The core AMPLIFY Portable communication backbone is based on a network-based publish-subscribe mechanism. This allows every node in the network to operate independently from one another, without having to have any knowledge of the other devices present in the network. Every input node can publish any number of data streams and output nodes can subscribe to the number of streams required to perform its output function.
This architecture is based around a central node which relays messages and streams and manages the publish-subscribe channels. This node is present at a well-known location in the network and input/output devices connect to it upon joining the network. The central node also provides a method of discovering client nodes present in the network and more importantly services they provide in a service registry. Client nodes register channels they want to provide and the type of data these channels contain before starting to publish data on said channel. Should the channel become unavailable, e.g. whenever a client leaves the network, it should be removed from the service registry as well.
The architecture and specifically the central node described in the previous paragraph is reasonably generic, so that it can be implemented using any number of technologies. For the first deployment near future of AMPLIFY Portable, the functionality for the central node will be provided by a Redis server and its publish-subscribe features.
Redis is an open source in-memory key-value store with a low memory footprint and is designed for high throughput and low latency with a simple communication protocol. It also supports master-replica replication, making it possible to scale out the central node to multiple processes and/or hosts should the need for it arise.
To simplify interaction with the central Redis node and provide a coherent interface across different client implementations, interface libraries will be provided which clients can use to interact with the system. These libraries will be specific to the programming language used the implement the client and will be provided on a on-demand basis.
This section provides an example of how a client library might interact with the central node. Also note, that in general, each channel on the central node will have a single publisher and can have multiple subscribers.
NB: These examples pertains to the client interface library and most of the implementation details will be transparent to the actual client code.
Input nodes provide streams to the system, e.g. devices such as sensors.
- Client input node connects to the Redis central node at a well-known address.
- Client registers its details with the central node as a JSON data structure under a key of type hash meant for this purpose. This data structure might contain information such as device type (input, output or both), number of streams it provides.
- The client device gets ready for publishing data, it registers a new service with the central node under a key of type hash intended for this purpose. The description of the service contains the name of the stream and the type of data it will contain (numbers, boolean, ...) and whether it will be continuous or sporadic. Registering the service before starting to publish data also prevents naming conflicts with other services, as in that case the central node will simply reject the service registration.
- The client node can start publishing data to the chosen publish-subscribe channel.
Output nodes are devices that consume data by ingesting data from publish-subscribe channels. They can use this data for instance for rendering data to a screen.
- Client output node connects to the Redis central node at a well-known address.
- Client registers its details with the central node as a JSON data structure under a key of type hash meant for this purpose. This data structure might contain information such as device type (input, output or both).
- The client device gets ready for ingesting data. It queries the service registry for streams that are available in the system and learns about the type of data it can expect from them.
- It subscribes to the chosen streams by providing a callback function, which will be triggered every time the publisher associated with the channel publishes a new sample.
It is important to note here that the output device can subscribe to any number of channels needed to accomplish its function. The same is true for input devices, though in general, they only publish on a single main channel.
Further, nodes in the system may not be strictly input or output devices, but a combination of both. For instance, a node might ingest several raw data streams, process them in some way and republish the processed data on a new channel.
Whenever a client node decides to go offline and leave the network or stop providing a service, it has to unregister its services from the service registry and remove itself from the client registry before disconnecting.
Should a client become unresponsive or crash, the service and device registries will become outdated and potentially provide unreliable information. In order to prevent this from happening, a heartbeat mechanism should be implemented, in which clients refresh a timestamp at a certain frequency in the device registry with which other clients can gauge whether a client is still available. This mechanism potentially requires a cleanup process which removes stale clients from the registries.
This section defines the API that users of the interface library can program against. Every concrete implementation of the interface library should support these methods. Method signatures use a Typescript-like syntax for annotating types. These signatures are not necessarily authoritative, but act merely as pseudocode.
The signatures for these functions assume a procedural implementation. An object-oriented implementation can, in general, assume the first parameter to be the owner of the given method if it happens to be an object.
connect(
host: string,
type: 'input' | 'output' | 'io',
deviceInfo: any?
): Connection
This function acts as the main entry point for the application. It establishes
a connection to the Redis server at the host/port identified by the string
host. A device should also register itself as either an input device,
output device and general io if it can perform both functions. Optionally
this method also accepts a dictionary-like object containing device
information. The contents of this object are as of yet unspecified.
The function establishes the connection to the server and returns a Connection
object through which all future communication shall be conducted.
close(
connection: Connection
)
Closes the given connection to the server. This will also unregister all the streams provided by the caller and unsubscribe from all registered streams.
registerStream(
connection: Connection,
name: string,
type: 'continuous' | 'discrete',
dataType: 'number' | 'string' | 'boolean'
): Stream
Registers a new data stream with the Redis server. The caller specifies a name
for the stream, whether the stream is continuous or discrete and what kind of
data is contained in the stream. The function returns a Stream object which
can be used to actually publish the samples to the server.
unregisterStream(
connection: Connection,
stream: Stream,
)
Deregisters a stream from the server, indicating that no more messages will be published on the stream from this point forward.
publish(
stream: Stream,
sample: any
)
Publishes a new sample on a registered stream, taking the sample as argument. This sample can be of any type, but must be (JSON) serialisable.
getAvailableStreams(
connection: Connection
): Dictionary<string, Stream>
Returns a dictionary of with streams names as keys and streams as values, that are currently available on the server. The objects in the dictionary inform the caller about stream type and type of data contained within the stream.
getStream(
connection: Connection,
streamName: string
): Stream
Returns a stream with the given name from the server and returns a Stream object if the stream exists
subscribe(
stream: Stream,
callback: (sample: any) => void
)
Subscribes the caller to messages published on the given stream by the server. This function accepts a callback, which is invoked every time a sample is received with the received sample as argument.
unsubscribe(
stream: Stream
)
Unsubscribes the caller from messages published on the given stream.