NaviNIBS.util.ZMQConnector module

exception NaviNIBS.util.ZMQConnector.InvalidMessageError

Bases: Exception

class NaviNIBS.util.ZMQConnector.ZMQConnectorServer(obj, reqRepPort, pubSubPort=None, bindAddr='127.0.0.1', ctx=NOTHING, onOtherRequestReceived=None)

Bases: object

Parameters:

• _obj

• _reqRepPort

• _pubSubPort

• _bindAddr

• _ctx

• _onOtherRequestReceived

close()

publish(msg)

exception NaviNIBS.util.ZMQConnector.RemoteError

Bases: Exception

class NaviNIBS.util.ZMQConnector.ZMQConnectorClient(reqRepPort, pubSubPort=None, connAddr='127.0.0.1', ctx=NOTHING, actx=NOTHING, onMessagePublished=None, allowAsyncCalls=False, allowSyncCalls=True)

Bases: object

Parameters:

• _reqRepPort

• _pubSubPort

• _connAddr

• _ctx

• _actx

• onMessagePublished (Optional[Callable[[List[bytes]], None]])

• _allowAsyncCalls

• _allowSyncCalls

onMessagePublished: typing.Optional[typing.Callable[[List[bytes]], None]]

property connAddr

close(linger=0)

async send_async(msg)

Return type:

List[bytes]

send(msg)

Return type:

List[bytes]

async get_async(item, raw=False)

Return type:

Any

get(item, raw=False)

Return type:

Any

async set_async(key, value, raw=False)

set(key, value, raw=False)

async call_async(method, *args, **kwargs)

Asynchronously on the client, make a synchronous call on the server

call(method, *args, **kwargs)

async callAsync_async(method, *args, **kwargs)

” Asynchronously on the client, make an asynchronous call on the server

callAsync(method, *args, **kwargs)

Synchronously on the client, make an asynchronous call on the server

async ping_async(timeout=100, numTries=2)

Parameters:

• timeout – time in ms

• numTries

Returns:

None

Raises TimeoutError if ping fails

ping(timeout=100, numTries=2)

NaviNIBS.util.ZMQConnector.checkIfPortAvailable(port)

Return type:

bool

NaviNIBS.util.ZMQConnector.getNewPort(minAssignedPort=5000)

Return type:

int

class NaviNIBS.util.ZMQConnector.Test_ZMQConnector(methodName='runTest')

Bases: TestCase

class state

Bases: object

a: int

b: str

c: typing.List[bytes]

async respondAfterDelay(delay=1.0, toReturn=True)

test_large_async()

test_large_sync()

test_basic_async()

test_basic_sync()

test_overlapping_async()