MQTT

Syntax

connect(host, port,[QoS=0],[formatter],[batchSize=0],[username],[password],[sendbufSize=40960],[config])

Details

This function connects to an MQTT server/broker. It returns a connection object that can be explicitly called to close with the close function, or it can be automatically released when the reference count is 0.

Parameters

• host: A string indicating the IP address of MQTT server/broker.

• port: An integer indicating the port number of MQTT server/broker.

• Qos (optional): An integer indicating the quality of service with the default value of 0.

  • 0: at most once;
  • 1: at least once;
  • 2: only once.

• formatter: A function used to package published data in CSV or JSON formats. Currently supported functions are createJsonFormatter and createCsvFormatter.

• batchSize: An integer indicating the number of rows sent each time. When the content to be published is a table, it can be sent in batches.

• username: A string indicating the user name of the MQTT server/broker.

• password: A string indicating the password of the MQTT server/broker.

• sendbufSize: An integer indicating the size of the send buffer in bytes, with a default value of 40960.

  config: A dictionary whose keys are strings and values are of ANY type. It is used to specify the following additional options:

  • "recvBufSize": An integer indicating the size of the receive buffer in bytes, with a default value of 40960.
  • "clientID": A string specifying the publisher ID. If not specified, it will be automatically generated.

Examples

f = createJsonFormatter()
conn = mqtt::connect("test.mosquitto.org",1883,0,f,50)

Syntax

publish(conn,topic,obj)

Details

This function publishes messages to the MQTT server/broker.

Parameters

• conn: An object generated by function connect.
• topic: A string indicating the topic.
• obj: A string or vector/in-memory table indicating the content of the message to be published.

Examples

mqtt::publish(conn,"dolphindb/topic1","welcome")
mqtt::publish(conn,"devStr/sensor001",["hello world","welcome"])
t = table(take(0..99,50) as hardwareId ,take(now(),
		50) as ts,rand(20..41,50) as temperature,
		rand(50,50) as humidity,rand(500..1000,
		50) as voltage)
mqtt::publish(conn,"dolphindb/device",t)		

createPublisher(conn,topic,colNames,colTypes)

Details

This function creates an object that can publish messages by writing data to it.

Parameters

• conn: An object generated by function connect.
• topic: A string indicating the topic.
• colNames: A STRING vector indicating the column names of the published schema.
• colTypes: A vector indicating the column types of the published schema.

Examples

MyFormat = take("", 3)
MyFormat[2] = "0.000"
f = createCsvFormatter(MyFormat, ',', ';')
pubConn = mqtt::connect("127.0.0.1",1883,0,f,100)

colNames = [`ts, `hardwareId, `val]
colTypes = [TIMESTAMP, SYMBOL, INT]
publisher = mqtt::createPublisher(pubConn, "sensor/s001", colNames, colTypes)

// example 1: by appending data to 'publisher' to publish it, the 'tb' here must be a table
append!(publisher, tb)

// example 2: by using the 'insert into' SQL clause
insert into publisher values([2023.08.25 10:57:47.961, 2023.08.25 10:57:47.961], symbol([`bb,`cc]), [22,33])

// example 3: by setting handler=append!{publisher} when subscribing a streaming table
share streamTable(1000:0, `time`sym`val, [TIMESTAMP, SYMBOL, INT]) as trades
subscribeTable(tableName="trades", actionName="engine1", offset=0, handler=append!{publisher}, msgAsTable=true);

insert into trades values(2018.10.08T01:01:01.785,`dd,44)
insert into trades values(2018.10.08T01:01:02.125,`ee,55)
insert into trades values(2018.10.08T01:01:10.263,`ff,66)

Syntax

close(conn)

Details

This function disconnects from the MQTT server/broker.

Parameters

• conn: An object generated by function connect.

Examples

mqtt::close(conn)

Syntax

subscribe(host, port, topic, [parser], handler, [username], [password], [recvbufSize=20480],[config])

Details

This function subscribes to the MQTT server/broker and returns a connection object.

Parameters

• host: A string indicating the IP address of the MQTT server/broker.
• port: An integer indicating the port number of the MQTT server/broker.
• topic: A string indicating the subscription topic.
• parser: A function used to parse subscribed messages into CSV or JSON formats. Currently supported functions are createJsonParser and createCsvParser.
• handler: A function or a table used to process the subscribed data from the MQTT server/broker.
• username: A string indicating the user name of the MQTT server/broker.
• password: A string indicating the password of the MQTT server/broker.
• recvbufSize: An integer indicating the size of the receive buffer in bytes, with a default value of 40960.
• config: A dictionary whose keys are strings and values are of ANY type. It is used to specify the following additional options:
  • "sendBufSize": An integer indicating the size of the send buffer in bytes, with a default value of 40960.
  • "subscribeID": A string specifying the subscriber ID. If not specified, it will be automatically generated.
  • "asyncFlag": A Boolean value specifying whether to enable the asynchronous subscription mode.

Examples

p = createCsvParser([INT, TIMESTAMP, DOUBLE, DOUBLE,DOUBLE], ',', ';' )
sensorInfoTable = table( 10000:0,`deviceID`send_time`temperature`humidity`voltage,[INT, TIMESTAMP, DOUBLE, DOUBLE,DOUBLE])
conn = mqtt::subscribe("192.168.1.201",1883,"sensor/#",p,sensorInfoTable)

getSubscriberStat()    

Details

This function gets the information on all subscriptions and returns a table with the following columns:

• "subscriptionId": ID of a subscription.
• "user": the session user who created the subscription.
• "host": IP address for the MQTT server/broker.
• "port": Port number of the MQTT server/broker.
• "topic": the subscription topic.
• "createTimestamp": the time when the subscription was created.
• "receivedPackets": the number of messages received.

Examples

mqtt::getSubscriberStat()

Syntax

unsubcribe(subscription)  

Details

This function cancels the subscription to the MQTT server/broker.

Parameters

• subscription: The value returned by the subscribe function or the subscription ID returned by getSubscriberStat.

Examples

mqtt::unsubscribe(sub1) 
mqtt::unsubscribe("350555232l")    

Syntax

createCsvFormatter([format], [delimiter=','], [rowDelimiter=';'])

Details

This function creates a formatter function in CSV format.

Parameters

• format: A STRING vector.
• delimiter: A string indicating the separator between columns, and the default is ','.
• rowDelimiter: A string indicating the separator between the lines, and the default is ';'.

Examples

def createT(n) {
    return table(take([false, true], n) as bool, take('a'..'z', n) as char, take(short(-5..5), n) as short, take(-5..5, n) as int, take(-5..5, n) as long, take(2001.01.01..2010.01.01, n) as date, take(2001.01M..2010.01M, n) as month, take(time(now()), n) as time, take(minute(now()), n) as minute, take(second(now()), n) as second, take(datetime(now()), n) as datetime, take(now(), n) as timestamp, take(nanotime(now()), n) as nanotime, take(nanotimestamp(now()), n) as nanotimestamp, take(3.1415, n) as float, take(3.1415, n) as double, take(`AAPL`IBM, n) as string, take(`AAPL`IBM, n) as symbol)
}
t = createT(100)
MyFormat = take("", 18)
MyFormat[2] = "0.000"
MyFormat[5] = "yyyy.MM.dd"
f = mqtt::createCsvFormatter(MyFormat)
f(t)

Syntax

createCsvParser(schema, [delimiter=','], [rowDelimiter=';'])

Details

This function creates a parser function in CSV format.

Parameters

• schema: A vector indicating the data types of columns.
• delimiter: A string indicating the separator between columns, and the default is ','.
• rowDelimiter: A string indicating the separator between the lines, and the default is ';'.

Examples

def createT(n) {
    return table(take([false, true], n) as bool, take('a'..'z', n) as char, take(short(-5..5), n) as short, take(-5..5, n) as int, take(-5..5, n) as long, take(2001.01.01..2010.01.01, n) as date, take(2001.01M..2010.01M, n) as month, take(time(now()), n) as time, take(minute(now()), n) as minute, take(second(now()), n) as second, take(datetime(now()), n) as datetime, take(now(), n) as timestamp, take(nanotime(now()), n) as nanotime, take(nanotimestamp(now()), n) as nanotimestamp, take(3.1415, n) as float, take(3.1415, n) as double, take(`AAPL`IBM, n) as string, take(`AAPL`IBM, n) as symbol)
}
t = createT(100)
f = mqtt::createCsvFormatter([BOOL,CHAR,SHORT,INT,LONG,DATE,MONTH,TIME,MINUTE,SECOND,DATETIME,TIMESTAMP,NANOTIME,NANOTIMESTAMP,FLOAT,DOUBLE,STRING,SYMBOL])
s=f(t)
p = mqtt::createCsvParser([BOOL,CHAR,SHORT,INT,LONG,DATE,MONTH,TIME,MINUTE,SECOND,DATETIME,TIMESTAMP,NANOTIME,NANOTIMESTAMP,FLOAT,DOUBLE,STRING,SYMBOL])
p(s)

Syntax

createJsonFormatter()

Details

This function creates a formatter function in JSON format.

Parameters

None.

Examples

def createT(n) {
    return table(take([false, true], n) as bool, take('a'..'z', n) as char, take(short(-5..5), n) as short, take(-5..5, n) as int, take(-5..5, n) as long, take(2001.01.01..2010.01.01, n) as date, take(2001.01M..2010.01M, n) as month, take(time(now()), n) as time, take(minute(now()), n) as minute, take(second(now()), n) as second, take(datetime(now()), n) as datetime, take(now(), n) as timestamp, take(nanotime(now()), n) as nanotime, take(nanotimestamp(now()), n) as nanotimestamp, take(3.1415, n) as float, take(3.1415, n) as double, take(`AAPL`IBM, n) as string, take(`AAPL`IBM, n) as symbol)
}
t = createT(100)
f = mqtt::createJsonFormatter()
f(t)

createJsonParser(schema, colNames)

Details

This function creates a parser function in JSON format.

Parameters

• schema: A vector indicating the data types of columns.
• colNames: A string indicating the column names.

Examples

def createT(n) {
    return table(take([false, true], n) as bool, take('a'..'z', n) as char, take(short(-5..5), n) as short, take(-5..5, n) as int, take(-5..5, n) as long, take(2001.01.01..2010.01.01, n) as date, take(2001.01M..2010.01M, n) as month, take(time(now()), n) as time, take(minute(now()), n) as minute, take(second(now()), n) as second, take(datetime(now()), n) as datetime, take(now(), n) as timestamp, take(nanotime(now()), n) as nanotime, take(nanotimestamp(now()), n) as nanotimestamp, take(3.1415, n) as float, take(3.1415, n) as double, take(`AAPL`IBM, n) as string, take(`AAPL`IBM, n) as symbol)
}
t = createT(100)
f = mqtt::createJsonFormatter()
p = mqtt::createJsonParser([BOOL,CHAR,SHORT,INT,LONG,DATE,MONTH,TIME,MINUTE,SECOND,DATETIME,TIMESTAMP,NANOTIME,NANOTIMESTAMP,FLOAT,DOUBLE,STRING,SYMBOL],
`bool`char`short`int`long`date`month`time`minute`second`datetime`timestamp`nanotime`nanotimestamp`float`double`string`symbol)
s=f(t)
x=p(s)