COSP

Python implementation

Connection oriented session protocol

hat.drivers.cosp.mlog

module logger

Type

logging.Logger

hat.drivers.cosp.Data

Data

alias of Union[bytes, bytearray, memoryview]

class hat.drivers.cosp.Address(host, port)

Bases: tuple

Create new instance of Address(host, port)

host

host name

Type

str

port

TCP port

Type

int

class hat.drivers.cosp.ConnectionInfo(local_addr, local_tsel, local_ssel, remote_addr, remote_tsel, remote_ssel)

Bases: tuple

Create new instance of ConnectionInfo(local_addr, local_tsel, local_ssel, remote_addr, remote_tsel, remote_ssel)

local_addr

local address

Type

Address

local_ssel

local COSP selector

Type

Optional[int]

local_tsel

local COTP selector

Type

Optional[int]

remote_addr

remote address

Type

Address

remote_ssel

remote COSP selector

Type

Optional[int]

remote_tsel

remote COTP selector

Type

Optional[int]

hat.drivers.cosp.ValidateResult

Validate result

alias of Optional[Union[bytes, bytearray, memoryview]]

hat.drivers.cosp.ValidateCb

Validate callback

alias of Callable[[Union[bytes, bytearray, memoryview]], Union[bytes, bytearray, memoryview, None, Awaitable[Optional[Union[bytes, bytearray, memoryview]]]]]

hat.drivers.cosp.ConnectionCb

Connection callback

alias of Callable[[Connection], Union[None, Awaitable[None]]]

async hat.drivers.cosp.connect(addr, local_tsel=None, remote_tsel=None, local_ssel=None, remote_ssel=None, user_data=None)

Connect to COSP server

Parameters
  • addr (Address) – remote address

  • local_tsel (Optional[int]) – local COTP selector

  • remote_tsel (Optional[int]) – remote COTP selector

  • local_ssel (Optional[int]) – local COSP selector

  • remote_ssel (Optional[int]) – remote COSP selector

  • user_data (Optional[Data]) – connect request user data

Returns

Connection

async hat.drivers.cosp.listen(validate_cb, connection_cb, addr=Address(host='0.0.0.0', port=102))

Create COSP listening server

Parameters
  • validate_cb (ValidateCb) – callback function or coroutine called on new incomming connection request prior to creating new connection

  • connection_cb (ConnectionCb) – new connection callback

  • addr (Address) – local listening address

Returns

Server

class hat.drivers.cosp.Server

Bases: object

COSP listening server

For creating new server see listen()

property addresses

listening addresses

Type

List[Address]

property closed

closed future

Type

asyncio.Future

async async_close()

Close listening socket

Calling this method doesn’t close active incomming connections

class hat.drivers.cosp.Connection

Bases: object

COSP connection

For creating new connection see hat.drivers.cosp.connect()

property info

connection info

Type

ConnectionInfo

property conn_req_user_data

connect request’s user data

Type

Data

property conn_res_user_data

connect response’s user data

Type

Data

property closed

closed future

Type

asyncio.Future

async async_close(user_data=None)

Async close

Parameters

user_data (Optional[Data]) – finish message user data

async read()

Read data

Returns

Data

write(data)

Write data

Parameters

data (Data) – data