Streaming data

Fetching the stream of data

Streaming data is a constant stream of data. Examples are Twitter streams, online news articles, and sensor data streams. There are several ways to fetch streaming data. We can directly connect to the server providing the data (or via API) and create a loop calling to the server periodically. (See also references below)

HTTPAPI's

#!/usr/bin/python

import socket

HOST = 'rtd.hpwren.ucsd.edu' # The remote host
PORT = 12020                

def isOpen(ip,port):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.connect((ip, int(port)))
        d = s.recv(1024).decode('utf-8')
        print(d)
        return True
    except Exception as e:
        print(e)
        return False
    finally:
        s.close()

while True:
    print(isOpen(HOST, PORT))
    sleep(5)

https://github.com/fenna/twitter_analysis/blob/master/scrabe_twitter.py demonstrate the scraping of Twitter feeds

This is not a problem in the case of a constant stream of data coming from a sensor. But this is not very efficient in the case of a new scientific paper release. It is like walking to a store 20 times asking at the counter if the new release is out and returning empty-handed 19 times because no new release is out yet.

source: https://refactoring.guru/design-patterns/observer

Observer

In these cases, it is best practise to make use of an observer object that informs you when something happens on the data producer site. The idea is that you subscribe to a data producer object and the subscription takes care of data delivery in case of new data. You can write your own or make use of available libraries such as MQTT.

MQTT

MQTT (Message Queuing Telemetry Transport) is a protocol with a special publish/subscribe implementation designed for IoT applications. Devices are not directly talking to each other. Instead, communication is structured into topics and handled over a central server (broker). When retrieving sensor data you define a client that subscribes to broker topic, for instance 'outside temperature'. The producer server publishes the sensor data to the MQTT broker and the broker takes care of delivering the data to the client. A topic is a string that the broker uses to filter the messages that the client requests.

source: 8bitwork.com

Callback

A callback function is a very efficient way to wait for things to happen at the 'data producer' end. The function will be called whenever a new message is available. a recommended name for such a function ison_message()

import paho.mqtt.subscribe as subscribe

def on_message(client, userdata, message):
    print(f'{message.topic}, {message.payload.decode("utf-8"})
    
subscribe.callback(on_message,
                   topics="data/temp",
                   hostname="test.mosquitto.org")