...
| Code Block |
|---|
// Receiving elements
#PARSER PQL
#RUNQUERY
in = RECEIVE({
transport = 'MQTT',
protocol = 'simplecsv',
source = 'mqttsource',
datahandler = 'Tuple',
options = [
['topic','Test'],
['Broker','tcp://localhost:1883'],
['Client_ID','OdysseusReceiver']
],
schema = [['Message', 'String']]
}
)
// Sending elements
#PARSER PQL
#RUNQUERY
ping = TIMER({PERIOD = 1000, SOURCE = 'Timer'})
input = MAP({EXPRESSIONS = ['"Hallo " + toString(counter())']}, ping)
out = SENDER({
transport = 'MQTT',
protocol = 'simplecsv',
sink = 'mqttsink',
WRAPPER = 'GenericPush',
datahandler = 'Tuple',
options = [
['topic','Test'],
['Broker','tcp://localhost:1883'],
['Client_ID','Odysseus_SenderOdysseusSender']
]
}, input
) |
Remark: Protocol and data handler are just examples. MQTT can be used with any other protcol and data handler (e.g. KeyValue).:
| Code Block |
|---|
#PARSER PQL
#RUNQUERY
in = RECEIVE({
transport = 'MQTT',
protocol = 'JSON',
source = 'mqttsource',
datahandler = 'KeyValueObject',
options = [
['topic','Test'],
['Broker','tcp://localhost:1883'],
['Client_ID','OdysseusReceiverKV']
]
}
)
#PARSER PQL
#RUNQUERY
ping = TIMER({PERIOD = 1000, SOURCE = 'Timer'})
input = MAP({EXPRESSIONS = [['"Hallo " + toString(counter())','message']]}, ping)
kv_input = TUPLETOKEYVALUE(input)
out = SENDER({
transport = 'MQTT',
protocol = 'JSON',
sink = 'mqttsink',
WRAPPER = 'GenericPush',
datahandler = 'KeyValueObject',
options = [
['topic','Test'],
['Broker','tcp://localhost:1883'],
['Client_ID','OdysseusSenderKV']
]
}, kv_input
)
|