<commit_before>ctypedef struct WriteContext:
    uv.uv_write_t req
    uv.uv_buf_t uv_buf
    Py_buffer py_buf
    PyObject* callback
    PyObject* transport
    Py_ssize_t len


cdef class UVTCPBase(UVStream):
    def __cinit__(self, Loop loop, *_):
        cdef int err

        self.handle = <uv.uv_handle_t*> \
                            PyMem_Malloc(sizeof(uv.uv_tcp_t))
        if self.handle is NULL:
            raise MemoryError()

        self.handle.data = <void*> self

        err = uv.uv_tcp_init(loop.loop, <uv.uv_tcp_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

        self.opened = 0

    cdef enable_nodelay(self):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_nodelay(<uv.uv_tcp_t *>self.handle, 1)
        if err < 0:
            raise UVError.from_error(err)

    cdef disable_nodelay(self):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_nodelay(<uv.uv_tcp_t *>self.handle, 0)
        if err < 0:
            raise UVError.from_error(err)


cdef class UVTCPServer(UVTCPBase):
    def __cinit__(self, *_):
        self.protocol_factory = None

    cdef set_protocol_factory(self, object protocol_factory):
        if self.protocol_factory is not None:
            raise RuntimeError('can only set protocol_factory once')
        self.protocol_factory = protocol_factory

    cdef open(self, int sockfd):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_open(<uv.uv_tcp_t *>self.handle, sockfd)
        if err < 0:
            raise UVError.from_error(err)
        self.opened = 1

    cdef bind(self, uv.sockaddr* addr, unsigned int flags=0):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_bind(<uv.uv_tcp_t *>self.handle,
                             addr, flags)
        if err < 0:
            raise UVError.from_error(err)
        self.opened = 1

    cdef listen(self, int backlog=100):
        cdef int err
        self.ensure_alive()

        if self.protocol_factory is None:
            raise RuntimeError('unable to listen(); no protocol_factory')

        if self.opened != 1:
            raise RuntimeError('unopened UVTCPServer')

        err = uv.uv_listen(<uv.uv_stream_t*> self.handle,
                           backlog,
                           __server_listen_cb)
        if err < 0:
            raise UVError.from_error(err)

    cdef _new_client(self):
        protocol = self.protocol_factory()

        client = UVServerTransport(self.loop, self)
        client._accept()
        self.loop._track_handle(client) # XXX


cdef class UVServerTransport(UVTCPBase):
    def __cinit__(self, Loop loop, UVTCPServer server):
        self.server = server

    cdef _accept(self):
        cdef int err
        self.ensure_alive()

        err = uv.uv_accept(<uv.uv_stream_t*>self.server.handle,
                           <uv.uv_stream_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

        self.opened = 1
        self._start_reading()

    cdef _start_reading(self):
        cdef int err
        self.ensure_alive()

        if self.opened != 1:
            raise RuntimeError(
                'cannot UVServerTransport.start_reading; opened=0')

        err = uv.uv_read_start(<uv.uv_stream_t*>self.handle,
                               __alloc_cb,
                               __server_transport_onread_cb)
        if err < 0:
            raise UVError.from_error(err)

    cdef _stop_reading(self):
        cdef int err
        self.ensure_alive()

        if self.opened != 1:
            raise RuntimeError(
                'cannot UVServerTransport.stop_reading; opened=0')

        err = uv.uv_read_stop(<uv.uv_stream_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

    cdef _on_data_recv(self, bytes buf):
        # print('>>>>>>>', buf)
        self._write(buf, None)

    cdef _on_eof(self):
        # print("EOF")
        self.close()

    cdef _write(self, object data, object callback):
        cdef:
            int err
            WriteContext* ctx

        ctx = <WriteContext*> PyMem_Malloc(sizeof(WriteContext))
        if ctx is NULL:
            raise MemoryError()

        try:<commit_msg>Check that handle is active in Transport._write<commit_after>ctypedef struct WriteContext:
    uv.uv_write_t req
    uv.uv_buf_t uv_buf
    Py_buffer py_buf
    PyObject* callback
    PyObject* transport
    Py_ssize_t len


cdef class UVTCPBase(UVStream):
    def __cinit__(self, Loop loop, *_):
        cdef int err

        self.handle = <uv.uv_handle_t*> \
                            PyMem_Malloc(sizeof(uv.uv_tcp_t))
        if self.handle is NULL:
            raise MemoryError()

        self.handle.data = <void*> self

        err = uv.uv_tcp_init(loop.loop, <uv.uv_tcp_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

        self.opened = 0

    cdef enable_nodelay(self):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_nodelay(<uv.uv_tcp_t *>self.handle, 1)
        if err < 0:
            raise UVError.from_error(err)

    cdef disable_nodelay(self):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_nodelay(<uv.uv_tcp_t *>self.handle, 0)
        if err < 0:
            raise UVError.from_error(err)


cdef class UVTCPServer(UVTCPBase):
    def __cinit__(self, *_):
        self.protocol_factory = None

    cdef set_protocol_factory(self, object protocol_factory):
        if self.protocol_factory is not None:
            raise RuntimeError('can only set protocol_factory once')
        self.protocol_factory = protocol_factory

    cdef open(self, int sockfd):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_open(<uv.uv_tcp_t *>self.handle, sockfd)
        if err < 0:
            raise UVError.from_error(err)
        self.opened = 1

    cdef bind(self, uv.sockaddr* addr, unsigned int flags=0):
        cdef int err
        self.ensure_alive()
        err = uv.uv_tcp_bind(<uv.uv_tcp_t *>self.handle,
                             addr, flags)
        if err < 0:
            raise UVError.from_error(err)
        self.opened = 1

    cdef listen(self, int backlog=100):
        cdef int err
        self.ensure_alive()

        if self.protocol_factory is None:
            raise RuntimeError('unable to listen(); no protocol_factory')

        if self.opened != 1:
            raise RuntimeError('unopened UVTCPServer')

        err = uv.uv_listen(<uv.uv_stream_t*> self.handle,
                           backlog,
                           __server_listen_cb)
        if err < 0:
            raise UVError.from_error(err)

    cdef _new_client(self):
        protocol = self.protocol_factory()

        client = UVServerTransport(self.loop, self)
        client._accept()
        self.loop._track_handle(client) # XXX


cdef class UVServerTransport(UVTCPBase):
    def __cinit__(self, Loop loop, UVTCPServer server):
        self.server = server

    cdef _accept(self):
        cdef int err
        self.ensure_alive()

        err = uv.uv_accept(<uv.uv_stream_t*>self.server.handle,
                           <uv.uv_stream_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

        self.opened = 1
        self._start_reading()

    cdef _start_reading(self):
        cdef int err
        self.ensure_alive()

        if self.opened != 1:
            raise RuntimeError(
                'cannot UVServerTransport.start_reading; opened=0')

        err = uv.uv_read_start(<uv.uv_stream_t*>self.handle,
                               __alloc_cb,
                               __server_transport_onread_cb)
        if err < 0:
            raise UVError.from_error(err)

    cdef _stop_reading(self):
        cdef int err
        self.ensure_alive()

        if self.opened != 1:
            raise RuntimeError(
                'cannot UVServerTransport.stop_reading; opened=0')

        err = uv.uv_read_stop(<uv.uv_stream_t*>self.handle)
        if err < 0:
            raise UVError.from_error(err)

    cdef _on_data_recv(self, bytes buf):
        # print('>>>>>>>', buf)
        self._write(buf, None)

    cdef _on_eof(self):
        # print("EOF")
        self.close()

    cdef _write(self, object data, object callback):
        cdef:
            int err
            WriteContext* ctx

        self.ensure_alive()

        ctx = <WriteContext*> PyMem_Malloc(sizeof(WriteContext))
        if ctx is NULL:
            raise MemoryError()

        try:
{'fraction_non_alphanumeric': 0.09193793022984482, 'fraction_numerical': 0.0056002800140007, 'mean_word_length': 1.8678487788558047, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '863672', 'n_tokens_mistral': 2817, 'n_tokens_neox': 2621, 'n_words': 594}