SocketWrapper

Complexity

Total Complexity

6

Size/Duplication

Total Lines	26
Duplicated Lines	0 %

Test Coverage

Coverage

88.24%

Importance

Changes	2
Bugs	0	Features	0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	write()	0	2	1
A	__init__()	0	2	1
A	read()	0	15	4

import logging
import os
from concurrent.futures import Future
import functools
import threading
from socket import error as SocketError

try:
    # we prefer to use bundles asyncio version, otherwise fallback to trollius
    import asyncio
except ImportError:

The variable ImportError does not seem to be defined.

    import trollius as asyncio


from opcua.common.uaerrors import UaError


class ServiceError(UaError):

The variable UaError does not seem to be defined.

    def __init__(self, code):
        super(ServiceError, self).__init__('UA Service Error')
        self.code = code

The variable code does not seem to be defined.

class NotEnoughData(UaError):

The variable UaError does not seem to be defined.

    pass


class SocketClosedException(UaError):
    pass


class Buffer(object):

The variable object does not seem to be defined.

    """
    alternative to io.BytesIO making debug easier
    and added a few conveniance methods
    """

    def __init__(self, data, start_pos=0, size=-1):
        # self.logger = logging.getLogger(__name__)
        self._data = data

The variable data does not seem to be defined.

        self._cur_pos = start_pos

The variable start_pos does not seem to be defined.

        if size == -1:

The variable size does not seem to be defined.

            size = len(data) - start_pos
        self._size = size

    def __str__(self):
        return "Buffer(size:{}, data:{})".format(
            self._size,

The variable self does not seem to be defined.

            self._data[self._cur_pos:self._cur_pos + self._size])
    __repr__ = __str__

    def __len__(self):
        return self._size

The variable self does not seem to be defined.

    def read(self, size):
        """
        read and pop number of bytes for buffer
        """
        if size > self._size:

The variable size does not seem to be defined.

            raise NotEnoughData("Not enough data left in buffer, request for {}, we have {}".format(size, self))

The variable self does not seem to be defined.

        # self.logger.debug("Request for %s bytes, from %s", size, self)
        self._size -= size
        pos = self._cur_pos
        self._cur_pos += size
        data = self._data[pos:self._cur_pos]

The variable pos does not seem to be defined.

        # self.logger.debug("Returning: %s ", data)
        return data

The variable data does not seem to be defined.

    def copy(self, size=-1):
        """
        return a shadow copy, optionnaly only copy 'size' bytes
        """
        if size == -1 or size > self._size:

The variable size does not seem to be defined.

            size = self._size

The variable self does not seem to be defined.

        return Buffer(self._data, self._cur_pos, size)

    def skip(self, size):
        """
        skip size bytes in buffer
        """
        if size > self._size:

The variable size does not seem to be defined.

            raise NotEnoughData("Not enough data left in buffer, request for {}, we have {}".format(size, self))

The variable self does not seem to be defined.

        self._size -= size
        self._cur_pos += size


class SocketWrapper(object):

The variable object does not seem to be defined.

    """
    wrapper to make it possible to have same api for 
    normal sockets, socket from asyncio, StringIO, etc....
    """
    def __init__(self, sock):
        self.socket = sock

The variable sock does not seem to be defined.

    def read(self, size):
        """
        Receive up to size bytes from socket
        """
        data = b''
        while size > 0:

The variable size does not seem to be defined.

            try:
                chunk = self.socket.recv(size)
            except (OSError, SocketError) as ex:

The variable OSError does not seem to be defined.

The variable SocketError does not seem to be defined.

                raise SocketClosedException("Server socket has closed", ex)

The variable ex does not seem to be defined.

            if not chunk:

The variable chunk does not seem to be defined.

                raise SocketClosedException("Server socket has closed")
            data += chunk

The variable data does not seem to be defined.

            size -= len(chunk)
        return data

    def write(self, data):
        self.socket.sendall(data)

The variable data does not seem to be defined.

def create_nonce(size=32):
    return os.urandom(size)


class ThreadLoop(threading.Thread):

The variable threading does not seem to be defined.

    """
    run an asyncio loop in a thread
    """

    def __init__(self):
        threading.Thread.__init__(self)

The variable self does not seem to be defined.

        self.logger = logging.getLogger(__name__)

The variable __name__ does not seem to be defined.

        self.loop = None
        self._cond = threading.Condition()

    def start(self):
        with self._cond:

The variable self does not seem to be defined.

            threading.Thread.start(self)
            self._cond.wait()

    def run(self):
        self.logger.debug("Starting subscription thread")
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)

The variable self does not seem to be defined.

        with self._cond:
            self._cond.notify_all()
        self.loop.run_forever()
        self.logger.debug("subscription thread ended")

    def create_server(self, proto, hostname, port):
        return self.loop.create_server(proto, hostname, port)

The variable proto does not seem to be defined.

The variable hostname does not seem to be defined.

The variable port does not seem to be defined.

    def stop(self):
        """
        stop subscription loop, thus the subscription thread
        """
        self.loop.call_soon_threadsafe(self.loop.stop)

The variable self does not seem to be defined.

    def call_soon(self, callback):
        self.loop.call_soon_threadsafe(callback)

The variable callback does not seem to be defined.

    def call_later(self, delay, callback):
        """
        threadsafe call_later from asyncio
        """
        p = functools.partial(self.loop.call_later, delay, callback)

The variable delay does not seem to be defined.

The variable self does not seem to be defined.

The variable callback does not seem to be defined.

        self.loop.call_soon_threadsafe(p)

The variable p does not seem to be defined.

    def _create_task(self, future, coro, cb=None):
        #task = self.loop.create_task(coro)
        task = asyncio.async(coro, loop=self.loop) 

The variable coro does not seem to be defined.

The variable self does not seem to be defined.

        if cb:

The variable cb does not seem to be defined.

            task.add_done_callback(cb)
        future.set_result(task)

The variable task does not seem to be defined.

    def create_task(self, coro, cb=None):
        """
        threadsafe create_task from asyncio
        """
        future = Future()
        p = functools.partial(self._create_task, future, coro, cb)

The variable cb does not seem to be defined.

The variable coro does not seem to be defined.

The variable future does not seem to be defined.

The variable self does not seem to be defined.

        self.loop.call_soon_threadsafe(p)

The variable p does not seem to be defined.

        return future.result()

    def run_coro_and_wait(self, coro):
        cond = threading.Condition()
        def cb(_):
            with cond:

The variable cond does not seem to be defined.

                cond.notify_all()
        with cond:
            task = self.create_task(coro, cb)

The variable coro does not seem to be defined.

The variable cb does not seem to be defined.

            cond.wait()
        return task.result()

    def _run_until_complete(self, future, coro):
        task = self.loop.run_until_complete(coro)

The variable coro does not seem to be defined.

        future.set_result(task)

The variable task does not seem to be defined.

    def run_until_complete(self, coro):
        """
        threadsafe run_until_completed from asyncio
        """
        future = Future()
        p = functools.partial(self._run_until_complete, future, coro)

The variable future does not seem to be defined.

The variable self does not seem to be defined.

The variable coro does not seem to be defined.

        self.loop.call_soon_threadsafe(p)

The variable p does not seem to be defined.

        return future.result()