threading -> asyncio queue

queue.py

import asyncio
import queue
import socket

class Error (Exception):
    pass

Full = queue.Full

class Queue:
    """
        An asyncio-thread queue using AF_UNIX sockets.

        A thread can put() arbitrary objects.

        These are placed on a threadsafe in-memory queue, and a single byte marker is written to a socketpair.

        An asyncio task can get() arbitrary objects, blocking until there is something available.
    """

    def __init__ (self):
        self.queue = queue.Queue()
        self.read, self.write = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)

        self.read.setblocking(False)
        self.write.setblocking(True)

    def put (self, msg):
        """
            Raises Full if the queue becomes full. Aye.
        """

        self.queue.put(msg)

        # signal
        if not self.write.send(b'.'):
            raise Full

    @asyncio.coroutine
    def get (self):
        loop = asyncio.get_event_loop()

        out = yield from loop.sock_recv(self.read, 1)

        try:
            msg = self.queue.get(block=False)
        except queue.Empty:
            raise Error("queue desync")

        return msg