A simple protocol for sending synchronous binary messages between two hosts.

_readme.md

Channel

A simple channel implementation for python for prototyping and debugging. It is thread-safe.

API

Channel.listen(<host>, <port>, reuse=True, echo=True) -> iterator over client channel objects

Args:

• <host>: The IP address to listen on.
• <port>: The TCP port to listen on.
• reuse=True: If set to true, it will set the SO_REUSEPORT socket flag (which you generally want while debugging).
• echo=False: If set to true, it will print basic information about connecting clients to STDOUT.

Return value: An iterator over the connecting clients. See below for an example.

Channel.connect(<host>, <port>) -> channel object

Args:

• <host>: The IP address to connect to.
• <port>: The TCP port to connect using.

Return value: A channel object connected to the given <host> and <port>.

Channel().send(<string / bytes>)

Sends the given string to the other end of the channel which can be received with an equivalent call to recv(). send()s are recv()d in-order.

Channel().recv() -> <string / bytes>

Receives the next string that was sent over this channel.

Example

On Server

for client_channel in Channel.listen("localhost", 5000, reuse=True):
  spin_off_client_thread(client_channel)

On Client

chan = Channel.connect("localhost", 5000)
chan.send("Hello World")
resp = chan.recv()
print resp

channel.py

import socket
from threading import Lock

### Outline of channel protocol
# To begin a message, the client or the server sends
#   <CR> <LF> "MSG" <SPACE> <[0-9]+> <CR> <LF> <bytes>
# That is to say:
#   - (<CR>) An ascii carriage return character.
#   - (<LF>) An ascii line-feed character.
#   - ("MSG") The literal string "MSG" (no quotes) in ascii.
#   - (<SPACE>) An ascii space character
#   - (<[0-9]+>) The number of bytes of the message body encodes as ascii numerals
#   - (<CR>) An ascii carriage return
#   - (<LF>) An ascii line-feed character
#   - (<bytes>) The bytes of the message. Must be the same length as the length
#     value sent before.

def recv_line(socket, chunk_size=1):
    buffer = ""
    r = socket.recv(chunk_size)
    while True:
        if not r: raise RuntimeError("Socket was closed while reading line")
        buffer += r
        if buffer[-1] == "\n":
            return buffer
        r = socket.recv(chunk_size)
    assert False, "Unreachable"

class Channel(object):

    @classmethod
    def connect(kls, ip, port):
        sock = socket.create_connection((ip, port))
        return kls(sock, ip, port)

    @classmethod
    def listen(kls, ip, port, reuse=True, echo=False):
        lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        if reuse:
            lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        lsock.bind((ip, port))
        lsock.listen(10)
        if echo: print "listening on {0}:{1}".format(ip, port)
        while True:
            csock, addr = lsock.accept()
            ip, port = addr
            if echo: print "--> {0}:{1}".format(ip, port)
            yield kls(csock, ip, port)

    def __init__(self, socket, ip, port):
        self.ip = ip
        self.port = port
        self.socket = socket
        self.w_lock = Lock()
        self.r_lock = Lock()

    def iter_recv(self):
        while True:
            yield self.recv()

    def send(self, value):
        value = bytes(value)
        msg = "\r\nMSG {0}\r\n".format(len(value)) + value
        try:
            self.w_lock.acquire()
            self.socket.sendall(msg)
        finally:
            self.w_lock.release()

    def recv(self):
        try:
            self.r_lock.acquire()
            crlf = self.socket.recv(2)
            line = recv_line(self.socket).strip()
            type, length = line.split(" ")
            assert type == "MSG"
            value = self.socket.recv(int(length))
            return value
        finally:
            self.r_lock.release()