Changeset: 0c61a0ebc2b9 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=0c61a0ebc2b9
Modified Files:
clients/iotapi/documentation/websockets_api.rst
clients/iotapi/src/Streams/streams.py
clients/iotapi/src/Streams/streamscontext.py
clients/iotapi/src/WebSockets/websockets.py
Branch: iot
Log Message:
Finished first version of documentation for web api. Added pattern for response
messages during a WebSocket session.
diffs (truncated from 401 to 300 lines):
diff --git a/clients/iotapi/documentation/websockets_api.rst
b/clients/iotapi/documentation/websockets_api.rst
--- a/clients/iotapi/documentation/websockets_api.rst
+++ b/clients/iotapi/documentation/websockets_api.rst
@@ -18,7 +18,7 @@ The following sections explain the avail
sub
---
-Subscribes for new basket creations from a specific stream. Whenever a basket
is created, the server sends a notification message indicating the number of
inserted tuples in the new basket. The user has to specify the stream's name
and schema.
+Subscribes for new basket creations from a specific stream. Whenever a basket
is created, the server sends a notification message indicating the number of
inserted tuples in the new basket. The user has to specify the stream's name
and schema. To subscribe to a temperature stream, the following would suffice:
.. code-block:: json
@@ -33,6 +33,14 @@ unsub
Unsubscribes a previous subscribed stream for a client. The user has to
specify the stream's name and schema. The example is the same as above, just
changing the request keyword.
+.. code-block:: json
+
+ {
+ "request": "unsub",
+ "schema": "measures",
+ "stream": "temperature"
+ }
+
read
----
@@ -40,7 +48,7 @@ Reads output result from baskets generat
.. important:: The user has not to be subscribed to the stream in order to
read data from it!
-It's possible to provide an offset, a limit and a basket number where the read
should start. The request will always provide a result, even if the query
provides no tuples to read. The user has to specify the stream's name and
schema.
+It's possible to provide an offset, a limit and a basket number where the read
should start. The request will always provide a result, even if the query
provides no tuples to read. The user has to specify the stream's name and
schema. A possible query for the temperatures stream:
.. code-block:: json
@@ -54,7 +62,7 @@ It's possible to provide an offset, a li
info
----
-Retrieves information about a giving stream if a stream's name and schema are
provided, or all the existing streams in the system otherwise.
+Retrieves information about a giving stream if a stream's name and schema are
provided, or all the existing streams in the system otherwise. To request
information of temperature stream:
.. code-block:: json
@@ -66,3 +74,178 @@ Retrieves information about a giving str
Responses
=========
+
+The response messages, are also provided with JSON. They contain a
:code:`response` field, identifying the type of the message.
+
+error
+-----
+
+An internal error occurred in the server during a client's request. The
:code:`message` contains a string explaining the error. The following error
message happens when the user attempts to unsubscribed to a non subscribed
stream:
+
+.. code-block:: json
+
+ {
+ "response": "error",
+ "message": "Stream measures.temperature not present in the user's
subscriptions!"
+ }
+
+subscribed
+----------
+
+Message confirming the subscription to new baskets notifications of a stream.
+
+.. code-block:: json
+
+ {
+ "response": "subscribed",
+ "schema": "measures",
+ "stream": "temperature"
+ }
+
+unsubscribed
+------------
+
+Message confirming the removal of a subscription to new baskets notifications
of a stream.
+
+.. code-block:: json
+
+ {
+ "response": "unsubscribed",
+ "schema": "measures",
+ "stream": "temperature"
+ }
+
+removed
+-------
+
+If a stream is removed on MonetDB's engine, but the client is still
subscribed, then this message is sent.
+
+.. code-block:: json
+
+ {
+ "response": "removed",
+ "schema": "measures",
+ "stream": "temperature"
+ }
+
+notification
+------------
+
+Notification of a new basket creation from a subscribed stream. The message
contains the basket number and the number of tuples in the new basket. A
possible notification example for the above stream:
+
+.. code-block:: json
+
+ {
+ "response": "notification",
+ "schema": "measures",
+ "stream": "temperature",
+ "basket": 2,
+ "count": 50
+ }
+
+read
+----
+
+Response message for a read query. Contains the reconstructed tuples
listening. The tuples are listened in the say they are inserted in the RESTful
webserver. If a column has a null value, the JSON's :code:`null` value will be
listened. A possible query result for the above stream:
+
+.. code-block:: json
+
+ {
+ "response": "read",
+ "schema": "measures",
+ "stream": "temperature",
+ "count": 3,
+ "tuples": [
+ {
+ "sensorid": "living room",
+ "temperature": 32.6
+ },
+ {
+ "sensorid": "kitchen",
+ "temperature": 34.2
+ },
+ {
+ "sensorid": "bathroom",
+ "temperature": 28.9
+ }
+ ]
+ }
+
+info
+----
+
+Message with details about a stream including both columns and baskets
details. Note that the possible types list are restricted to the MonetDB
kernel. A possible example for the above stream:
+
+.. code-block:: json
+
+ {
+ "response": "info",
+ "schema": "measures",
+ "stream": "temperature",
+ "columns": [
+ {
+ "name": "sensorid",
+ "type": "clob"
+ },
+ {
+ "name": "temperature",
+ "type": "real"
+ }
+ ],
+ "baskets_count": 3,
+ "baskets_listing": [
+ {
+ "number": 1,
+ "count": 20
+ },
+ {
+ "number": 2,
+ "count": 25
+ },
+ {
+ "number": 3,
+ "count": 12
+ }
+ ]
+ }
+
+data
+----
+
+Return a info messaging regarding all the streams in the system. An example
with the temperatures stream:
+
+.. code-block:: json
+
+ {
+ "response": "data",
+ "streams_count": 1,
+ "streams_listing": [
+ "schema": "measures",
+ "stream": "temperature",
+ "columns": [
+ {
+ "name": "sensorid",
+ "type": "clob"
+ },
+ {
+ "name": "temperature",
+ "type": "real"
+ }
+ ],
+ "baskets_count": 3,
+ "baskets_listing": [
+ {
+ "number": 1,
+ "count": 20
+ },
+ {
+ "number": 2,
+ "count": 25
+ },
+ {
+ "number": 3,
+ "count": 12
+ }
+ ]
+ ]
+ }
diff --git a/clients/iotapi/src/Streams/streams.py
b/clients/iotapi/src/Streams/streams.py
--- a/clients/iotapi/src/Streams/streams.py
+++ b/clients/iotapi/src/Streams/streams.py
@@ -31,7 +31,8 @@ class StreamBasketsHandler(FileSystemEve
if isinstance(event, DirCreatedEvent):
basket_string = os.path.basename(os.path.normpath(event.src_path))
count = self._stream.append_basket(basket_string)
- notify_stream_inserts_to_clients(self._stream.get_schema_name(),
self._stream.get_stream_name(), count)
+ notify_stream_inserts_to_clients(self._stream.get_schema_name(),
self._stream.get_stream_name(),
+ int(basket_string), count)
def on_deleted(self, event):
if isinstance(event, DirDeletedEvent):
@@ -65,10 +66,11 @@ class IOTStream(object):
dic = OrderedDict({'schema': self._schema_name, 'stream':
self._stream_name,
'columns': [value.to_json_representation() for
value in self._columns.values()]})
self._baskets_lock.acquire_read()
- baskets = {'count': len(self._baskets),
- 'details': [{'number': k, 'total': v} for k, v in
self._baskets.items()]}
+ count = len(self._baskets)
+ listing = [{'number': k, 'count': v} for k, v in self._baskets.items()]
self._baskets_lock.release()
- dic['baskets'] = baskets
+ dic['baskets_count'] = count
+ dic['baskets_listing'] = listing
return dic
def append_basket(self, path):
@@ -142,4 +144,4 @@ class IOTStream(object):
keys = results.keys() # TODO check if this is viable for many tuples!!
tuples = [dict(zip(keys, values)) for values in zip(*(results[k] for k
in keys))]
- return {'total': read_tuples, 'tuples': tuples}
+ return {'schema': self._schema_name, 'stream': self._stream_name,
'count': read_tuples, 'tuples': tuples}
diff --git a/clients/iotapi/src/Streams/streamscontext.py
b/clients/iotapi/src/Streams/streamscontext.py
--- a/clients/iotapi/src/Streams/streamscontext.py
+++ b/clients/iotapi/src/Streams/streamscontext.py
@@ -39,8 +39,8 @@ class IOTStreams(object):
def get_streams_data(self):
self._locker.acquire_read()
- res = {'count': len(self._context),
- 'details': [value.get_data_dictionary() for value in
self._context.values()]}
+ res = {'streams_count': len(self._context),
+ 'streams_listing': [value.get_data_dictionary() for value in
self._context.values()]}
self._locker.release()
return res
diff --git a/clients/iotapi/src/WebSockets/websockets.py
b/clients/iotapi/src/WebSockets/websockets.py
--- a/clients/iotapi/src/WebSockets/websockets.py
+++ b/clients/iotapi/src/WebSockets/websockets.py
@@ -26,11 +26,10 @@ def unsubscribe_removed_streams(concaten
from Streams.streamscontext import Streams_Context, IOTStreams # avoid
circular dependency
-def notify_stream_inserts_to_clients(schema_name, stream_name, count):
- concatenated_name = IOTStreams.get_context_entry_name(schema_name,
stream_name)
+def notify_stream_inserts_to_clients(schema_name, stream_name, basket_number,
count):
WebClientsLock.acquire_read()
for client in WebClients:
- client.send_notification_message(concatenated_name, schema_name,
stream_name, count)
+ client.send_notification_message(schema_name, stream_name,
basket_number, count)
WebClientsLock.release()
@@ -43,8 +42,8 @@ class IOTAPI(WebSocket):
self._subscriptions_locker = RWLock()
def sendJSONMessage(self, response, message): # IMPORTANT always use this
method to send messages to clients!!!!!
- json_message = json.dumps({'response': response, 'message': message})
- super(IOTAPI, self).sendMessage(json_message) # send JSON Strings to
clients
+ message['response'] = response
+ super(IOTAPI, self).sendMessage(json.dumps(message)) # send JSON
Strings to clients
def handleConnected(self): # overriden
WebClientsLock.acquire_write()
_______________________________________________
checkin-list mailing list
checkin-list@monetdb.org
https://www.monetdb.org/mailman/listinfo/checkin-list
