nghttp2.pyx

# nghttp2 - HTTP/2 C Library

# Copyright (c) 2013 Tatsuhiro Tsujikawa

# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:

# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.

# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
cimport cnghttp2

from libc.stdlib cimport malloc, free
from libc.string cimport memcpy, memset
from libc.stdint cimport uint8_t, uint16_t, uint32_t, int32_t
import logging


DEFAULT_HEADER_TABLE_SIZE = cnghttp2.NGHTTP2_DEFAULT_HEADER_TABLE_SIZE
DEFLATE_MAX_HEADER_TABLE_SIZE = 4096

HD_ENTRY_OVERHEAD = 32

class HDTableEntry:

    def __init__(self, name, namelen, value, valuelen):
        self.name = name
        self.namelen = namelen
        self.value = value
        self.valuelen = valuelen

    def space(self):
        return self.namelen + self.valuelen + HD_ENTRY_OVERHEAD

cdef _get_pybytes(uint8_t *b, uint16_t blen):
    return b[:blen]

cdef class HDDeflater:
    '''Performs header compression. The constructor takes
    |hd_table_bufsize_max| parameter, which limits the usage of header
    table in the given amount of bytes. This is necessary because the
    header compressor and decompressor share the same amount of
    header table and the decompressor decides that number. The
    compressor may not want to use all header table size because of
    limited memory availability. In that case, the
    |hd_table_bufsize_max| can be used to cap the upper limit of table
    size whatever the header table size is chosen by the decompressor.
    The default value of |hd_table_bufsize_max| is 4096 bytes.

    The following example shows how to compress request header sets:

        import binascii, nghttp2

        deflater = nghttp2.HDDeflater()
        res = deflater.deflate([(b'foo', b'bar'),
                              (b'baz', b'buz')])
        print(binascii.b2a_hex(res))

    '''

    cdef cnghttp2.nghttp2_hd_deflater *_deflater

    def __cinit__(self, hd_table_bufsize_max = DEFLATE_MAX_HEADER_TABLE_SIZE):
        rv = cnghttp2.nghttp2_hd_deflate_new(&self._deflater,
                                             hd_table_bufsize_max)
        if rv != 0:
            raise Exception(_strerror(rv))

    def __dealloc__(self):
        cnghttp2.nghttp2_hd_deflate_del(self._deflater)

    def deflate(self, headers):
        '''Compresses the |headers|. The |headers| must be sequence of tuple
        of name/value pair, which are sequence of bytes (not unicode
        string).

        This function returns the encoded header block in byte string.
        An exception will be raised on error.

        '''
        cdef cnghttp2.nghttp2_nv *nva = <cnghttp2.nghttp2_nv*>\
                                        malloc(sizeof(cnghttp2.nghttp2_nv)*\
                                        len(headers))
        cdef cnghttp2.nghttp2_nv *nvap = nva

        for k, v in headers:
            nvap[0].name = k
            nvap[0].namelen = len(k)
            nvap[0].value = v
            nvap[0].valuelen = len(v)
            nvap[0].flags = cnghttp2.NGHTTP2_NV_FLAG_NONE
            nvap += 1

        cdef size_t outcap = 0
        cdef ssize_t rv
        cdef uint8_t *out
        cdef size_t outlen

        outlen = cnghttp2.nghttp2_hd_deflate_bound(self._deflater,
                                                   nva, len(headers))

        out = <uint8_t*>malloc(outlen)

        rv = cnghttp2.nghttp2_hd_deflate_hd(self._deflater, out, outlen,
                                            nva, len(headers))
        free(nva)

        if rv < 0:
            free(out)

            raise Exception(_strerror(rv))

        cdef bytes res

        try:
            res = out[:rv]
        finally:
            free(out)

        return res

    def change_table_size(self, hd_table_bufsize_max):
        '''Changes header table size to |hd_table_bufsize_max| byte.

        An exception will be raised on error.

        '''
        cdef int rv
        rv = cnghttp2.nghttp2_hd_deflate_change_table_size(self._deflater,
                                                           hd_table_bufsize_max)
        if rv != 0:
            raise Exception(_strerror(rv))

    def get_hd_table(self):
        '''Returns copy of current dynamic header table.'''
        cdef size_t length = cnghttp2.nghttp2_hd_deflate_get_num_table_entries(
            self._deflater)
        cdef const cnghttp2.nghttp2_nv *nv
        res = []
        for i in range(62, length + 1):
            nv = cnghttp2.nghttp2_hd_deflate_get_table_entry(self._deflater, i)
            k = _get_pybytes(nv.name, nv.namelen)
            v = _get_pybytes(nv.value, nv.valuelen)
            res.append(HDTableEntry(k, nv.namelen, v, nv.valuelen))
        return res

cdef class HDInflater:
    '''Performs header decompression.

    The following example shows how to compress request header sets:

        data = b'0082c5ad82bd0f000362617a0362757a'
        inflater = nghttp2.HDInflater()
        hdrs = inflater.inflate(data)
        print(hdrs)

    '''

    cdef cnghttp2.nghttp2_hd_inflater *_inflater

    def __cinit__(self):
        rv = cnghttp2.nghttp2_hd_inflate_new(&self._inflater)
        if rv != 0:
            raise Exception(_strerror(rv))

    def __dealloc__(self):
        cnghttp2.nghttp2_hd_inflate_del(self._inflater)

    def inflate(self, data):
        '''Decompresses the compressed header block |data|. The |data| must be
        byte string (not unicode string).

        '''
        cdef cnghttp2.nghttp2_nv nv
        cdef int inflate_flags
        cdef ssize_t rv
        cdef uint8_t *buf = data
        cdef size_t buflen = len(data)
        res = []
        while True:
            inflate_flags = 0
            rv = cnghttp2.nghttp2_hd_inflate_hd2(self._inflater, &nv,
                                                 &inflate_flags,
                                                 buf, buflen, 1)
            if rv < 0:
                raise Exception(_strerror(rv))
            buf += rv
            buflen -= rv
            if inflate_flags & cnghttp2.NGHTTP2_HD_INFLATE_EMIT:
                # may throw
                res.append((nv.name[:nv.namelen], nv.value[:nv.valuelen]))
            if inflate_flags & cnghttp2.NGHTTP2_HD_INFLATE_FINAL:
                break

        cnghttp2.nghttp2_hd_inflate_end_headers(self._inflater)
        return res

    def change_table_size(self, hd_table_bufsize_max):
        '''Changes header table size to |hd_table_bufsize_max| byte.

        An exception will be raised on error.

        '''
        cdef int rv
        rv = cnghttp2.nghttp2_hd_inflate_change_table_size(self._inflater,
                                                           hd_table_bufsize_max)
        if rv != 0:
            raise Exception(_strerror(rv))

    def get_hd_table(self):
        '''Returns copy of current dynamic header table.'''
        cdef size_t length = cnghttp2.nghttp2_hd_inflate_get_num_table_entries(
            self._inflater)
        cdef const cnghttp2.nghttp2_nv *nv
        res = []
        for i in range(62, length + 1):
            nv = cnghttp2.nghttp2_hd_inflate_get_table_entry(self._inflater, i)
            k = _get_pybytes(nv.name, nv.namelen)
            v = _get_pybytes(nv.value, nv.valuelen)
            res.append(HDTableEntry(k, nv.namelen, v, nv.valuelen))
        return res

cdef _strerror(int liberror_code):
    return cnghttp2.nghttp2_strerror(liberror_code).decode('utf-8')

def print_hd_table(hdtable):
    '''Convenient function to print |hdtable| to the standard output. This
    function does not work if header name/value cannot be decoded using
    UTF-8 encoding.

    s=N means the entry occupies N bytes in header table.

    '''
    idx = 0
    for entry in hdtable:
        idx += 1
        print('[{}] (s={}) {}: {}'\
              .format(idx, entry.space(),
                      entry.name.decode('utf-8'),
                      entry.value.decode('utf-8')))

try:
    import socket
    import io
    import asyncio
    import traceback
    import sys
    import email.utils
    import datetime
    import time
    import ssl as tls
    from urllib.parse import urlparse
except ImportError:
    asyncio = None

# body generator flags
DATA_OK = 0
DATA_EOF = 1
DATA_DEFERRED = 2

class _ByteIOWrapper:

    def __init__(self, b):
        self.b = b

    def generate(self, n):
        data = self.b.read1(n)
        if not data:
            return None, DATA_EOF
        return data, DATA_OK

def wrap_body(body):
    if body is None:
        return body
    elif isinstance(body, str):
        return _ByteIOWrapper(io.BytesIO(body.encode('utf-8'))).generate
    elif isinstance(body, bytes):
        return _ByteIOWrapper(io.BytesIO(body)).generate
    elif isinstance(body, io.IOBase):
        return _ByteIOWrapper(body).generate
    else:
        # assume that callable in the form f(n) returning tuple byte
        # string and flag.
        return body

def negotiated_protocol(ssl_obj):
    protocol = ssl_obj.selected_alpn_protocol()
    if protocol:
        logging.info('alpn, protocol:%s', protocol)
        return protocol

    protocol = ssl_obj.selected_npn_protocol()
    if protocol:
        logging.info('npn, protocol:%s', protocol)
        return protocol

    return None

def set_application_protocol(ssl_ctx):
    app_protos = [cnghttp2.NGHTTP2_PROTO_VERSION_ID.decode('utf-8')]
    ssl_ctx.set_npn_protocols(app_protos)
    if tls.HAS_ALPN:
        ssl_ctx.set_alpn_protocols(app_protos)

cdef _get_stream_user_data(cnghttp2.nghttp2_session *session,
                           int32_t stream_id):
    cdef void *stream_user_data

    stream_user_data = cnghttp2.nghttp2_session_get_stream_user_data\
                       (session, stream_id)
    if stream_user_data == NULL:
        return None

    return <object>stream_user_data

cdef size_t _make_nva(cnghttp2.nghttp2_nv **nva_ptr, headers):
    cdef cnghttp2.nghttp2_nv *nva
    cdef size_t nvlen

    nvlen = len(headers)
    nva = <cnghttp2.nghttp2_nv*>malloc(sizeof(cnghttp2.nghttp2_nv) * nvlen)
    for i, (k, v) in enumerate(headers):
        nva[i].name = k
        nva[i].namelen = len(k)
        nva[i].value = v
        nva[i].valuelen = len(v)
        nva[i].flags = cnghttp2.NGHTTP2_NV_FLAG_NONE

    nva_ptr[0] = nva

    return nvlen

cdef int server_on_header(cnghttp2.nghttp2_session *session,
                          const cnghttp2.nghttp2_frame *frame,
                          const uint8_t *name, size_t namelen,
                          const uint8_t *value, size_t valuelen,
                          uint8_t flags,
                          void *user_data):
    cdef http2 = <_HTTP2SessionCoreBase>user_data
    logging.debug('server_on_header, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    handler = _get_stream_user_data(session, frame.hd.stream_id)
    return on_header(name, namelen, value, valuelen, flags, handler)

cdef int client_on_header(cnghttp2.nghttp2_session *session,
                          const cnghttp2.nghttp2_frame *frame,
                          const uint8_t *name, size_t namelen,
                          const uint8_t *value, size_t valuelen,
                          uint8_t flags,
                          void *user_data):
    cdef http2 = <_HTTP2SessionCoreBase>user_data
    logging.debug('client_on_header, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
        handler = _get_stream_user_data(session, frame.hd.stream_id)
    elif frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
        handler = _get_stream_user_data(session, frame.push_promise.promised_stream_id)

    return on_header(name, namelen, value, valuelen, flags, handler)


cdef int on_header(const uint8_t *name, size_t namelen,
                          const uint8_t *value, size_t valuelen,
                          uint8_t flags,
                          object handler):
    if not handler:
        return 0

    key = name[:namelen]
    values = value[:valuelen].split(b'\x00')
    if key == b':scheme':
        handler.scheme = values[0]
    elif key == b':method':
        handler.method = values[0]
    elif key == b':authority' or key == b'host':
        handler.host = values[0]
    elif key == b':path':
        handler.path = values[0]
    elif key == b':status':
        handler.status = values[0]

    if key == b'cookie':
        handler.cookies.extend(values)
    else:
        for v in values:
            handler.headers.append((key, v))

    return 0

cdef int server_on_begin_request_headers(cnghttp2.nghttp2_session *session,
                                         const cnghttp2.nghttp2_frame *frame,
                                         void *user_data):
    cdef http2 = <_HTTP2SessionCore>user_data

    handler = http2._make_handler(frame.hd.stream_id)
    cnghttp2.nghttp2_session_set_stream_user_data(session, frame.hd.stream_id,
                                                  <void*>handler)

    return 0

cdef int server_on_begin_headers(cnghttp2.nghttp2_session *session,
                                 const cnghttp2.nghttp2_frame *frame,
                                 void *user_data):
    if frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
        if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_REQUEST:
            return server_on_begin_request_headers(session, frame, user_data)

    return 0

cdef int server_on_frame_recv(cnghttp2.nghttp2_session *session,
                              const cnghttp2.nghttp2_frame *frame,
                              void *user_data):
    cdef http2 = <_HTTP2SessionCore>user_data
    logging.debug('server_on_frame_recv, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_DATA:
        if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
            handler = _get_stream_user_data(session, frame.hd.stream_id)
            if not handler:
                return 0
            try:
                handler.on_request_done()
            except:
                sys.stderr.write(traceback.format_exc())
                return http2._rst_stream(frame.hd.stream_id)
    elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
        if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_REQUEST:
            handler = _get_stream_user_data(session, frame.hd.stream_id)
            if not handler:
                return 0
            if handler.cookies:
                handler.headers.append((b'cookie',
                                        b'; '.join(handler.cookies)))
                handler.cookies = None
            try:
                handler.on_headers()
                if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
                    handler.on_request_done()
            except:
                sys.stderr.write(traceback.format_exc())
                return http2._rst_stream(frame.hd.stream_id)
    elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
        if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK):
            http2._stop_settings_timer()

    return 0

cdef int on_data_chunk_recv(cnghttp2.nghttp2_session *session,
                                   uint8_t flags,
                                   int32_t stream_id, const uint8_t *data,
                                   size_t length, void *user_data):
    cdef http2 = <_HTTP2SessionCoreBase>user_data

    handler = _get_stream_user_data(session, stream_id)
    if not handler:
        return 0

    try:
        handler.on_data(data[:length])
    except:
        sys.stderr.write(traceback.format_exc())
        return http2._rst_stream(stream_id)

    return 0

cdef int server_on_frame_send(cnghttp2.nghttp2_session *session,
                              const cnghttp2.nghttp2_frame *frame,
                              void *user_data):
    cdef http2 = <_HTTP2SessionCore>user_data
    logging.debug('server_on_frame_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
        # For PUSH_PROMISE, send push response immediately
        handler = _get_stream_user_data\
                  (session, frame.push_promise.promised_stream_id)
        if not handler:
            return 0

        http2.send_response(handler)
    elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
        if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK) != 0:
            return 0
        http2._start_settings_timer()
    elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
        if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM) and \
           cnghttp2.nghttp2_session_check_server_session(session):
            # Send RST_STREAM if remote is not closed yet
            if cnghttp2.nghttp2_session_get_stream_remote_close(
                    session, frame.hd.stream_id) == 0:
                http2._rst_stream(frame.hd.stream_id, cnghttp2.NGHTTP2_NO_ERROR)

cdef int server_on_frame_not_send(cnghttp2.nghttp2_session *session,
                                  const cnghttp2.nghttp2_frame *frame,
                                  int lib_error_code,
                                  void *user_data):
    cdef http2 = <_HTTP2SessionCore>user_data
    logging.debug('server_on_frame_not_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
        # We have to remove handler here. Without this, it is not
        # removed until session is terminated.
        handler = _get_stream_user_data\
                  (session, frame.push_promise.promised_stream_id)
        if not handler:
            return 0
        http2._remove_handler(handler)

cdef int on_stream_close(cnghttp2.nghttp2_session *session,
                                int32_t stream_id,
                                uint32_t error_code,
                                void *user_data):
    cdef http2 = <_HTTP2SessionCoreBase>user_data
    logging.debug('on_stream_close, stream_id:%s', stream_id)

    handler = _get_stream_user_data(session, stream_id)
    if not handler:
        return 0

    try:
        handler.on_close(error_code)
    except:
        sys.stderr.write(traceback.format_exc())

    http2._remove_handler(handler)

    return 0

cdef ssize_t data_source_read(cnghttp2.nghttp2_session *session,
                              int32_t stream_id,
                              uint8_t *buf, size_t length,
                              uint32_t *data_flags,
                              cnghttp2.nghttp2_data_source *source,
                              void *user_data):
    cdef http2 = <_HTTP2SessionCoreBase>user_data
    generator = <object>source.ptr

    http2.enter_callback()
    try:
        data, flag = generator(length)
    except:
        sys.stderr.write(traceback.format_exc())
        return cnghttp2.NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
    finally:
        http2.leave_callback()

    if flag == DATA_DEFERRED:
        return cnghttp2.NGHTTP2_ERR_DEFERRED

    if data:
        nread = len(data)
        memcpy(buf, <uint8_t*>data, nread)
    else:
        nread = 0

    if flag == DATA_EOF:
        data_flags[0] = cnghttp2.NGHTTP2_DATA_FLAG_EOF
        if cnghttp2.nghttp2_session_check_server_session(session):
            # Send RST_STREAM if remote is not closed yet
            if cnghttp2.nghttp2_session_get_stream_remote_close(
                    session, stream_id) == 0:
                http2._rst_stream(stream_id, cnghttp2.NGHTTP2_NO_ERROR)
    elif flag != DATA_OK:
        return cnghttp2.NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE

    return nread

cdef int client_on_begin_headers(cnghttp2.nghttp2_session *session,
                                 const cnghttp2.nghttp2_frame *frame,
                                 void *user_data):
    cdef http2 = <_HTTP2ClientSessionCore>user_data

    if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
        # Generate a temporary handler until the headers are all received
        push_handler = BaseResponseHandler()
        http2._add_handler(push_handler, frame.push_promise.promised_stream_id)
        cnghttp2.nghttp2_session_set_stream_user_data(session, frame.push_promise.promised_stream_id,
                                                      <void*>push_handler)

    return 0

cdef int client_on_frame_recv(cnghttp2.nghttp2_session *session,
                              const cnghttp2.nghttp2_frame *frame,
                              void *user_data):
    cdef http2 = <_HTTP2ClientSessionCore>user_data
    logging.debug('client_on_frame_recv, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_DATA:
        if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
            handler = _get_stream_user_data(session, frame.hd.stream_id)
            if not handler:
                return 0
            try:
                handler.on_response_done()
            except:
                sys.stderr.write(traceback.format_exc())
                return http2._rst_stream(frame.hd.stream_id)
    elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
        if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_RESPONSE or frame.headers.cat == cnghttp2.NGHTTP2_HCAT_PUSH_RESPONSE:
            handler = _get_stream_user_data(session, frame.hd.stream_id)

            if not handler:
                return 0
            # TODO handle 1xx non-final response
            if handler.cookies:
                handler.headers.append((b'cookie',
                                        b'; '.join(handler.cookies)))
                handler.cookies = None
            try:
                handler.on_headers()
                if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
                    handler.on_response_done()
            except:
                sys.stderr.write(traceback.format_exc())
                return http2._rst_stream(frame.hd.stream_id)
    elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
        if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK):
            http2._stop_settings_timer()
    elif frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
        handler = _get_stream_user_data(session, frame.hd.stream_id)
        if not handler:
            return 0
        # Get the temporary push_handler which now should have all of the header data
        push_handler = _get_stream_user_data(session, frame.push_promise.promised_stream_id)
        if not push_handler:
            return 0
        # Remove the temporary handler
        http2._remove_handler(push_handler)
        cnghttp2.nghttp2_session_set_stream_user_data(session, frame.push_promise.promised_stream_id,
                                                      <void*>NULL)

        try:
            handler.on_push_promise(push_handler)
        except:
            sys.stderr.write(traceback.format_exc())
            return http2._rst_stream(frame.hd.stream_id)

    return 0

cdef int client_on_frame_send(cnghttp2.nghttp2_session *session,
                              const cnghttp2.nghttp2_frame *frame,
                              void *user_data):
    cdef http2 = <_HTTP2ClientSessionCore>user_data
    logging.debug('client_on_frame_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)

    if frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
        if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK) != 0:
            return 0
        http2._start_settings_timer()

cdef class _HTTP2SessionCoreBase:
    cdef cnghttp2.nghttp2_session *session
    cdef transport
    cdef handler_class
    cdef handlers
    cdef settings_timer
    cdef inside_callback

    def __cinit__(self, transport, handler_class=None):
        self.session = NULL
        self.transport = transport
        self.handler_class = handler_class
        self.handlers = set()
        self.settings_timer = None
        self.inside_callback = False

    def __dealloc__(self):
        cnghttp2.nghttp2_session_del(self.session)

    def data_received(self, data):
        cdef ssize_t rv

        rv = cnghttp2.nghttp2_session_mem_recv(self.session, data, len(data))
        if rv < 0:
            raise Exception('nghttp2_session_mem_recv failed: {}'.format\
                            (_strerror(rv)))
        self.send_data()

    OUTBUF_MAX = 65535
    SETTINGS_TIMEOUT = 5.0

    def send_data(self):
        cdef ssize_t outbuflen
        cdef const uint8_t *outbuf

        while True:
            if self.transport.get_write_buffer_size() > self.OUTBUF_MAX:
                break
            outbuflen = cnghttp2.nghttp2_session_mem_send(self.session, &outbuf)
            if outbuflen == 0:
                break
            if outbuflen < 0:
                raise Exception('nghttp2_session_mem_send faild: {}'.format\
                                (_strerror(outbuflen)))
            self.transport.write(outbuf[:outbuflen])

        if self.transport.get_write_buffer_size() == 0 and \
           cnghttp2.nghttp2_session_want_read(self.session) == 0 and \
           cnghttp2.nghttp2_session_want_write(self.session) == 0:
            self.transport.close()

    def resume(self, stream_id):
        cnghttp2.nghttp2_session_resume_data(self.session, stream_id)
        if not self.inside_callback:
            self.send_data()

    def enter_callback(self):
        self.inside_callback = True

    def leave_callback(self):
        self.inside_callback = False

    def _make_handler(self, stream_id):
        logging.debug('_make_handler, stream_id:%s', stream_id)
        handler = self.handler_class(self, stream_id)
        self.handlers.add(handler)
        return handler

    def _remove_handler(self, handler):
        logging.debug('_remove_handler, stream_id:%s', handler.stream_id)
        self.handlers.remove(handler)

    def _add_handler(self, handler, stream_id):
        logging.debug('_add_handler, stream_id:%s', stream_id)
        handler.stream_id = stream_id
        handler.http2 = self
        handler.remote_address = self._get_remote_address()
        handler.client_certificate = self._get_client_certificate()
        self.handlers.add(handler)

    def _rst_stream(self, stream_id,
                   error_code=cnghttp2.NGHTTP2_INTERNAL_ERROR):
        cdef int rv

        rv = cnghttp2.nghttp2_submit_rst_stream\
             (self.session, cnghttp2.NGHTTP2_FLAG_NONE,
              stream_id, error_code)

        return rv

    def _get_remote_address(self):
        return self.transport.get_extra_info('peername')

    def _get_client_certificate(self):
        sock = self.transport.get_extra_info('socket')
        try:
            return sock.getpeercert()
        except AttributeError:
            return None

    def _start_settings_timer(self):
        loop = asyncio.get_event_loop()
        self.settings_timer = loop.call_later(self.SETTINGS_TIMEOUT,
                                              self._settings_timeout)

    def _stop_settings_timer(self):
        if self.settings_timer:
            self.settings_timer.cancel()
            self.settings_timer = None

    def _settings_timeout(self):
        cdef int rv

        logging.debug('_settings_timeout')

        self.settings_timer = None

        rv = cnghttp2.nghttp2_session_terminate_session\
             (self.session, cnghttp2.NGHTTP2_SETTINGS_TIMEOUT)
        try:
            self.send_data()
        except Exception as err:
            sys.stderr.write(traceback.format_exc())
            self.transport.close()
            return

    def _log_request(self, handler):
        now = datetime.datetime.now()
        tv = time.mktime(now.timetuple())
        datestr = email.utils.formatdate(timeval=tv, localtime=False,
                                        usegmt=True)
        try:
            method = handler.method.decode('utf-8')
        except:
            method = handler.method
        try:
            path = handler.path.decode('utf-8')
        except:
            path = handler.path
        logging.info('%s - - [%s] "%s %s HTTP/2" %s - %s', handler.remote_address[0],
                          datestr, method, path, handler.status,
                          'P' if handler.pushed else '-')

    def close(self):
        rv = cnghttp2.nghttp2_session_terminate_session\
             (self.session, cnghttp2.NGHTTP2_NO_ERROR)
        try:
            self.send_data()
        except Exception as err:
            sys.stderr.write(traceback.format_exc())
            self.transport.close()
            return

cdef class _HTTP2SessionCore(_HTTP2SessionCoreBase):
    def __cinit__(self, *args, **kwargs):
        cdef cnghttp2.nghttp2_session_callbacks *callbacks
        cdef cnghttp2.nghttp2_settings_entry iv[2]
        cdef int rv

        super(_HTTP2SessionCore, self).__init__(*args, **kwargs)

        rv = cnghttp2.nghttp2_session_callbacks_new(&callbacks)

        if rv != 0:
            raise Exception('nghttp2_session_callbacks_new failed: {}'.format\
                            (_strerror(rv)))

        cnghttp2.nghttp2_session_callbacks_set_on_header_callback(
            callbacks, server_on_header)
        cnghttp2.nghttp2_session_callbacks_set_on_begin_headers_callback(
            callbacks, server_on_begin_headers)
        cnghttp2.nghttp2_session_callbacks_set_on_frame_recv_callback(
            callbacks, server_on_frame_recv)
        cnghttp2.nghttp2_session_callbacks_set_on_stream_close_callback(
            callbacks, on_stream_close)
        cnghttp2.nghttp2_session_callbacks_set_on_frame_send_callback(
            callbacks, server_on_frame_send)
        cnghttp2.nghttp2_session_callbacks_set_on_frame_not_send_callback(
            callbacks, server_on_frame_not_send)
        cnghttp2.nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
            callbacks, on_data_chunk_recv)

        rv = cnghttp2.nghttp2_session_server_new(&self.session, callbacks,
                                                 <void*>self)

        cnghttp2.nghttp2_session_callbacks_del(callbacks)

        if rv != 0:
            raise Exception('nghttp2_session_server_new failed: {}'.format\
                            (_strerror(rv)))

        iv[0].settings_id = cnghttp2.NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS
        iv[0].value = 100
        iv[1].settings_id = cnghttp2.NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE
        iv[1].value = cnghttp2.NGHTTP2_INITIAL_WINDOW_SIZE

        rv = cnghttp2.nghttp2_submit_settings(self.session,
                                              cnghttp2.NGHTTP2_FLAG_NONE,
                                              iv, sizeof(iv) / sizeof(iv[0]))

        if rv != 0:
            raise Exception('nghttp2_submit_settings failed: {}'.format\
                            (_strerror(rv)))

    def send_response(self, handler):
        cdef cnghttp2.nghttp2_data_provider prd
        cdef cnghttp2.nghttp2_data_provider *prd_ptr
        cdef cnghttp2.nghttp2_nv *nva
        cdef size_t nvlen
        cdef int rv

        logging.debug('send_response, stream_id:%s', handler.stream_id)

        nva = NULL
        nvlen = _make_nva(&nva, handler.response_headers)

        if handler.response_body:
            prd.source.ptr = <void*>handler.response_body
            prd.read_callback = data_source_read
            prd_ptr = &prd
        else:
            prd_ptr = NULL

        rv = cnghttp2.nghttp2_submit_response(self.session, handler.stream_id,
                                              nva, nvlen, prd_ptr)

        free(nva)

        if rv != 0:
            # TODO Ignore return value
            self._rst_stream(handler.stream_id)
            raise Exception('nghttp2_submit_response failed: {}'.format\
                            (_strerror(rv)))

        self._log_request(handler)

    def push(self, handler, promised_handler):
        cdef cnghttp2.nghttp2_nv *nva
        cdef size_t nvlen
        cdef int32_t promised_stream_id

        self.handlers.add(promised_handler)

        nva = NULL
        nvlen = _make_nva(&nva, promised_handler.headers)

        promised_stream_id = cnghttp2.nghttp2_submit_push_promise\
                             (self.session,
                              cnghttp2.NGHTTP2_FLAG_NONE,
                              handler.stream_id,
                              nva, nvlen,
                              <void*>promised_handler)
        if promised_stream_id < 0:
            raise Exception('nghttp2_submit_push_promise failed: {}'.format\
                            (_strerror(promised_stream_id)))

        promised_handler.stream_id = promised_stream_id

        logging.debug('push, stream_id:%s', promised_stream_id)

        return promised_handler

    def connection_lost(self):
        self._stop_settings_timer()

        for handler in self.handlers:
            handler.on_close(cnghttp2.NGHTTP2_INTERNAL_ERROR)
        self.handlers = set()

cdef class _HTTP2ClientSessionCore(_HTTP2SessionCoreBase):
    def __cinit__(self, *args, **kwargs):
        cdef cnghttp2.nghttp2_session_callbacks *callbacks
        cdef cnghttp2.nghttp2_settings_entry iv[2]
        cdef int rv

        super(_HTTP2ClientSessionCore, self).__init__(*args, **kwargs)

        rv = cnghttp2.nghttp2_session_callbacks_new(&callbacks)

        if rv != 0:
            raise Exception('nghttp2_session_callbacks_new failed: {}'.format\
                            (_strerror(rv)))

        cnghttp2.nghttp2_session_callbacks_set_on_header_callback(
            callbacks, client_on_header)
        cnghttp2.nghttp2_session_callbacks_set_on_begin_headers_callback(
            callbacks, client_on_begin_headers)
        cnghttp2.nghttp2_session_callbacks_set_on_frame_recv_callback(
            callbacks, client_on_frame_recv)
        cnghttp2.nghttp2_session_callbacks_set_on_stream_close_callback(
            callbacks, on_stream_close)
        cnghttp2.nghttp2_session_callbacks_set_on_frame_send_callback(
            callbacks, client_on_frame_send)
        cnghttp2.nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
            callbacks, on_data_chunk_recv)

        rv = cnghttp2.nghttp2_session_client_new(&self.session, callbacks,
                                                 <void*>self)

        cnghttp2.nghttp2_session_callbacks_del(callbacks)

        if rv != 0:
            raise Exception('nghttp2_session_client_new failed: {}'.format\
                            (_strerror(rv)))

        iv[0].settings_id = cnghttp2.NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS
        iv[0].value = 100
        iv[1].settings_id = cnghttp2.NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE
        iv[1].value = cnghttp2.NGHTTP2_INITIAL_WINDOW_SIZE

        rv = cnghttp2.nghttp2_submit_settings(self.session,
                                              cnghttp2.NGHTTP2_FLAG_NONE,
                                              iv, sizeof(iv) / sizeof(iv[0]))

        if rv != 0:
            raise Exception('nghttp2_submit_settings failed: {}'.format\
                            (_strerror(rv)))

    def send_request(self, method, scheme, host, path, headers, body, handler):
        cdef cnghttp2.nghttp2_data_provider prd
        cdef cnghttp2.nghttp2_data_provider *prd_ptr
        cdef cnghttp2.nghttp2_priority_spec *pri_ptr
        cdef cnghttp2.nghttp2_nv *nva
        cdef size_t nvlen
        cdef int32_t stream_id

        body = wrap_body(body)

        custom_headers = _encode_headers(headers)
        headers = [
            (b':method', method.encode('utf-8')),
            (b':scheme', scheme.encode('utf-8')),
            (b':authority', host.encode('utf-8')),
            (b':path', path.encode('utf-8'))
        ]
        headers.extend(custom_headers)

        nva = NULL
        nvlen = _make_nva(&nva, headers)

        if body:
            prd.source.ptr = <void*>body
            prd.read_callback = data_source_read
            prd_ptr = &prd
        else:
            prd_ptr = NULL

        # TODO: Enable priorities
        pri_ptr = NULL

        stream_id = cnghttp2.nghttp2_submit_request\
                             (self.session, pri_ptr,
                              nva, nvlen, prd_ptr,
                              <void*>handler)
        free(nva)

        if stream_id < 0:
            raise Exception('nghttp2_submit_request failed: {}'.format\
                            (_strerror(stream_id)))

        logging.debug('request, stream_id:%s', stream_id)

        self._add_handler(handler, stream_id)
        cnghttp2.nghttp2_session_set_stream_user_data(self.session, stream_id,
                                                  <void*>handler)

        return handler

    def push(self, push_promise, handler):
        if handler:
            # push_promise accepted, fill in the handler with the stored
            # headers from the push_promise
            handler.status = push_promise.status
            handler.scheme = push_promise.scheme
            handler.method = push_promise.method
            handler.host = push_promise.host
            handler.path = push_promise.path
            handler.cookies = push_promise.cookies
            handler.stream_id = push_promise.stream_id
            handler.http2 = self
            handler.pushed = True

            self._add_handler(handler, handler.stream_id)

            cnghttp2.nghttp2_session_set_stream_user_data(self.session, handler.stream_id,
                                                      <void*>handler)
        else:
            # push_promise rejected, reset the stream
            self._rst_stream(push_promise.stream_id,
                              error_code=cnghttp2.NGHTTP2_NO_ERROR)

if asyncio:

    class BaseRequestHandler:

        """HTTP/2 request (stream) handler base class.

        The class is used to handle the HTTP/2 stream. By default, it does
        not nothing. It must be subclassed to handle each event callback
        method.

        The first callback method invoked is on_headers(). It is called
        when HEADERS frame, which includes request header fields, is
        arrived.

        If request has request body, on_data(data) is invoked for each
        chunk of received data.

        When whole request is received, on_request_done() is invoked.

        When stream is closed, on_close(error_code) is called.

        The application can send response using send_response() method. It
        can be used in on_headers(), on_data() or on_request_done().

        The application can push resource using push() method. It must be
        used before send_response() call.

        The following instance variables are available:

        client_address
          Contains a tuple of the form (host, port) referring to the client's
          address.

        client_certificate
          May contain the client certifcate in its non-binary form

        stream_id
          Stream ID of this stream

        scheme
          Scheme of the request URI. This is a value of :scheme header field.

        method
          Method of this stream. This is a value of :method header field.

        host
          This is a value of :authority or host header field.

        path
          This is a value of :path header field.

        headers
          Request header fields

        """

        def __init__(self, http2, stream_id):
            self.headers = []
            self.cookies = []
            # Stream ID. For promised stream, it is initially -1.
            self.stream_id = stream_id
            self.http2 = http2
            # address of the client
            self.remote_address = self.http2._get_remote_address()
            # certificate of the client
            self._client_certificate = self.http2._get_client_certificate()
            # :scheme header field in request
            self.scheme = None
            # :method header field in request
            self.method = None
            # :authority or host header field in request
            self.host = None
            # :path header field in request
            self.path = None
            # HTTP status
            self.status = None
            # True if this is a handler for pushed resource
            self.pushed = False

        @property
        def client_address(self):
            return self.remote_address

        @property
        def client_certificate(self):
            return self._client_certificate

        def on_headers(self):

            '''Called when request HEADERS is arrived.

            '''
            pass

        def on_data(self, data):

            '''Called when a chunk of request body is arrived. This method
            will be called multiple times until all data are received.

            '''
            pass

        def on_request_done(self):

            '''Called when whole request was received

            '''
            pass

        def on_close(self, error_code):

            '''Called when stream is about to close.

            '''
            pass

        def send_response(self, status=200, headers=None, body=None):

            '''Send response. The status is HTTP status code. The headers is
            additional response headers. The :status header field is
            appended by the library. The body is the response body. It
            could be None if response body is empty. Or it must be
            instance of either str, bytes, io.IOBase or callable,
            called body generator, which takes one parameter,
            size. The body generator generates response body. It can
            pause generation of response so that it can wait for slow
            backend data generation. When invoked, it should return
            tuple, byte string and flag. The flag is either DATA_OK,
            DATA_EOF and DATA_DEFERRED. For non-empty byte string and
            it is not the last chunk of response, DATA_OK is returned
            as flag.  If this is the last chunk of the response (byte
            string is possibly None), DATA_EOF must be returned as
            flag.  If there is no data available right now, but
            additional data are anticipated, return tuple (None,
            DATA_DEFERRD).  When data arrived, call resume() and
            restart response body transmission.

            Only the body generator can pause response body
            generation; instance of io.IOBase must not block.

            If instance of str is specified as body, it is encoded
            using UTF-8.

            The headers is a list of tuple of the form (name,
            value). The name and value can be either unicode string or
            byte string.

            On error, exception will be thrown.

            '''
            if self.status is not None:
                raise Exception('response has already been sent')

            if not status:
                raise Exception('status must not be empty')

            body = wrap_body(body)

            self._set_response_prop(status, headers, body)
            self.http2.send_response(self)

        def push(self, path, method='GET', request_headers=None,
                 status=200, headers=None, body=None):

            '''Push a resource. The path is a path portion of request URI
            for this
            resource. The method is a method to access this
            resource. The request_headers is additional request
            headers to access this resource. The :scheme, :method,
            :authority and :path are appended by the library. The
            :scheme and :authority are inherited from the request (not
            request_headers parameter).

            The status is HTTP status code. The headers is additional
            response headers. The :status header field is appended by
            the library. The body is the response body. It has the
            same semantics of body parameter of send_response().

            The headers and request_headers are a list of tuple of the
            form (name, value). The name and value can be either
            unicode string or byte string.

            On error, exception will be thrown.

            '''
            if not status:
                raise Exception('status must not be empty')

            if not method:
                raise Exception('method must not be empty')

            if not path:
                raise Exception('path must not be empty')

            body = wrap_body(body)

            promised_handler = self.http2._make_handler(-1)
            promised_handler.pushed = True
            promised_handler.scheme = self.scheme
            promised_handler.method = method.encode('utf-8')
            promised_handler.host = self.host
            promised_handler.path = path.encode('utf-8')
            promised_handler._set_response_prop(status, headers, body)

            headers = [
                (b':method', promised_handler.method),
                (b':scheme', promised_handler.scheme),
                (b':authority', promised_handler.host),
                (b':path', promised_handler.path)
            ]
            headers.extend(_encode_headers(request_headers))

            promised_handler.headers = headers

            return self.http2.push(self, promised_handler)

        def _set_response_prop(self, status, headers, body):
            self.status = status

            if headers is None:
                headers = []

            self.response_headers = [(b':status', str(status).encode('utf-8'))]
            self.response_headers.extend(_encode_headers(headers))

            self.response_body = body

        def resume(self):
            self.http2.resume(self.stream_id)

    def _encode_headers(headers):
        if not headers:
            return []
        return [(k if isinstance(k, bytes) else k.encode('utf-8'),
                 v if isinstance(v, bytes) else v.encode('utf-8')) \
                for k, v in headers]

    class _HTTP2Session(asyncio.Protocol):

        def __init__(self, RequestHandlerClass):
            asyncio.Protocol.__init__(self)
            self.RequestHandlerClass = RequestHandlerClass
            self.http2 = None

        def connection_made(self, transport):
            address = transport.get_extra_info('peername')
            logging.info('connection_made, address:%s, port:%s', address[0], address[1])

            self.transport = transport
            sock = self.transport.get_extra_info('socket')
            try:
                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            except OSError as e:
                logging.info('failed to set tcp-nodelay: %s', str(e))
            ssl_ctx = self.transport.get_extra_info('sslcontext')
            if ssl_ctx:
                ssl_obj = self.transport.get_extra_info('ssl_object')
                protocol = negotiated_protocol(ssl_obj)
                if protocol is None or protocol.encode('utf-8') != \
                   cnghttp2.NGHTTP2_PROTO_VERSION_ID:
                    self.transport.abort()
                    return
            try:
                self.http2 = _HTTP2SessionCore\
                             (self.transport,
                              self.RequestHandlerClass)
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.abort()
                return


        def connection_lost(self, exc):
            logging.info('connection_lost')
            if self.http2:
                self.http2.connection_lost()
                self.http2 = None

        def data_received(self, data):
            try:
                self.http2.data_received(data)
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.close()
                return

        def resume_writing(self):
            try:
                self.http2.send_data()
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.close()
                return

    class HTTP2Server:

        '''HTTP/2 server.

        This class builds on top of the asyncio event loop. On
        construction, RequestHandlerClass must be given, which must be a
        subclass of BaseRequestHandler class.

        '''
        def __init__(self, address, RequestHandlerClass, ssl=None):

            '''address is a tuple of the listening address and port (e.g.,
            ('127.0.0.1', 8080)). RequestHandlerClass must be a subclass
            of BaseRequestHandler class to handle a HTTP/2 stream.  The
            ssl can be ssl.SSLContext instance. If it is not None, the
            resulting server is SSL/TLS capable.

            '''
            def session_factory():
                return _HTTP2Session(RequestHandlerClass)

            self.loop = asyncio.get_event_loop()

            if ssl:
                set_application_protocol(ssl)

            coro = self.loop.create_server(session_factory,
                                           host=address[0], port=address[1],
                                           ssl=ssl)
            self.server = self.loop.run_until_complete(coro)
            logging.info('listen, address:%s, port:%s', address[0], address[1])

        def serve_forever(self):
            try:
                self.loop.run_forever()
            finally:
                self.server.close()
                self.loop.close()



    class BaseResponseHandler:

        """HTTP/2 response (stream) handler base class.

        The class is used to handle the HTTP/2 stream. By default, it does
        not nothing. It must be subclassed to handle each event callback
        method.

        The first callback method invoked is on_headers(). It is called
        when HEADERS frame, which includes response header fields, is
        arrived.

        If response has a body, on_data(data) is invoked for each
        chunk of received data.

        When whole response is received, on_response_done() is invoked.

        When stream is closed or underlying connection is lost,
        on_close(error_code) is called.

        The application can send follow up requests using HTTP2Client.send_request() method.

        The application can handle push resource using on_push_promise() method.

        The following instance variables are available:

        server_address
          Contains a tuple of the form (host, port) referring to the server's
          address.

        stream_id
          Stream ID of this stream

        scheme
          Scheme of the request URI. This is a value of :scheme header field.

        method
          Method of this stream. This is a value of :method header field.

        host
          This is a value of :authority or host header field.

        path
          This is a value of :path header field.

        headers
          Response header fields.  There is a special exception.  If this
          object is passed to push_promise(), this instance variable contains
          pushed request header fields.

        """

        def __init__(self, http2=None, stream_id=-1):
            self.headers = []
            self.cookies = []
            # Stream ID. For promised stream, it is initially -1.
            self.stream_id = stream_id
            self.http2 = http2
            # address of the server
            self.remote_address = None
            # :scheme header field in request
            self.scheme = None
            # :method header field in request
            self.method = None
            # :authority or host header field in request
            self.host = None
            # :path header field in request
            self.path = None
            # HTTP status
            self.status = None
            # True if this is a handler for pushed resource
            self.pushed = False

        @property
        def server_address(self):
            return self.remote_address

        def on_headers(self):

            '''Called when response HEADERS is arrived.

            '''
            pass

        def on_data(self, data):

            '''Called when a chunk of response body is arrived. This method
            will be called multiple times until all data are received.

            '''
            pass

        def on_response_done(self):

            '''Called when whole response was received

            '''
            pass

        def on_close(self, error_code):

            '''Called when stream is about to close.

            '''
            pass

        def on_push_promise(self, push_promise):

            '''Called when a push is promised. Default behavior is to
            cancel the push. If application overrides this method,
            it should call either accept_push or reject_push.

            '''
            self.reject_push(push_promise)

        def reject_push(self, push_promise):

            '''Convenience method equivalent to calling accept_push
            with a falsy value.

            '''
            self.http2.push(push_promise, None)

        def accept_push(self, push_promise, handler=None):

            '''Accept a push_promise and provider a handler for the
            new stream. If a falsy value is supplied for the handler,
            the push is rejected.

            '''
            self.http2.push(push_promise, handler)

        def resume(self):
            self.http2.resume(self.stream_id)

    class _HTTP2ClientSession(asyncio.Protocol):

        def __init__(self, client):
            asyncio.Protocol.__init__(self)
            self.http2 = None
            self.pending = []
            self.client = client

        def connection_made(self, transport):
            address = transport.get_extra_info('peername')
            logging.info('connection_made, address:%s, port:%s', address[0], address[1])

            self.transport = transport
            sock = self.transport.get_extra_info('socket')
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            ssl_ctx = self.transport.get_extra_info('sslcontext')
            if ssl_ctx:
                ssl_obj = self.transport.get_extra_info('ssl_object')
                protocol = negotiated_protocol(ssl_obj)
                if protocol is None or protocol.encode('utf-8') != \
                   cnghttp2.NGHTTP2_PROTO_VERSION_ID:
                    self.transport.abort()

            self.http2 = _HTTP2ClientSessionCore(self.transport)

	    # Clear pending requests
            send_pending = self.pending
            self.pending = []
            for method,scheme,host,path,headers,body,handler in send_pending:
                self.send_request(method=method, scheme=scheme, host=host, path=path,\
                                  headers=headers, body=body, handler=handler)
            self.http2.send_data()

        def connection_lost(self, exc):
            logging.info('connection_lost')
            if self.http2:
                self.http2 = None
            self.client.close()

        def data_received(self, data):
            try:
                self.http2.data_received(data)
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.close()
                return

        def resume_writing(self):
            try:
                self.http2.send_data()
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.close()
                return

        def send_request(self, method, scheme, host, path, headers, body, handler):
            try:
		# Waiting until connection established
                if not self.http2:
                    self.pending.append([method, scheme, host, path, headers, body, handler])
                    return

                self.http2.send_request(method=method, scheme=scheme, host=host, path=path,\
                                        headers=headers, body=body, handler=handler)
                self.http2.send_data()
            except Exception as err:
                sys.stderr.write(traceback.format_exc())
                self.transport.close()
                return

        def close(self):
            if self.http2:
                self.http2.close()


    class HTTP2Client:

        '''HTTP/2 client.

        This class builds on top of the asyncio event loop.

        '''
        def __init__(self, address, loop=None, ssl=None):

            '''address is a tuple of the connect address and port (e.g.,
            ('127.0.0.1', 8080)). The ssl can be ssl.SSLContext instance.
            If it is not None, the resulting client is SSL/TLS capable.
            '''

            self.address = address
            self.session = _HTTP2ClientSession(self)
            def session_factory():
                return self.session

            if ssl:
                set_application_protocol(ssl)

            self.loop = loop
            if not self.loop:
                self.loop = asyncio.get_event_loop()

            coro = self.loop.create_connection(session_factory,
                                           host=address[0], port=address[1],
                                           ssl=ssl)

            if ssl:
                self.scheme = 'https'
            else:
                self.scheme = 'http'

            self.transport,_ = self.loop.run_until_complete(coro)
            logging.info('connect, address:%s, port:%s', self.address[0], self.address[1])

        @property
        def io_loop(self):
            return self.loop

        def close(self):
            self.session.close()

        def send_request(self, method='GET', url='/', headers=None, body=None, handler=None):
            url = urlparse(url)
            scheme = url.scheme if url.scheme else self.scheme
            host = url.netloc if url.netloc else self.address[0]+':'+str(self.address[1])
            path = url.path
            if url.params:
                path += ';'+url.params
            if url.query:
                path += '?'+url.query
            if url.fragment:
                path += '#'+url.fragment

            self.session.send_request(method=method, scheme=scheme, host=host, path=path,\
                                      headers=headers, body=body, handler=handler)