mqtt.golo

dsl.mqtt.golo.md

module iot_one

# Publishing

import littleIOT
import talkToMQTTBroker # it's a promise
import noticeable

import gololang.concurrent.workers.WorkerEnvironment


function main = |args| {

	let mybroker = broker()
		: protocol("tcp")
		: host("test.mosquitto.org")
		: port(1883)

	let options = littleIOT(): getConnectOptions()

	let iot = littleIOT(): broker(mybroker): connectOptions(options): initialize(): connect()

	iot
		: messageArrived(|topic, message| -> println("you've got a mail : " + topic + " | " + message) )
		: deliveryComplete(|token| -> println("delivery is complete " + token) )
		: connectionLost(|error| -> println("ouch"))
		: activateCallBacks()

	let env = WorkerEnvironment.builder(): withCachedThreadPool()

  # run promise
  talkToMQTTBroker(env, {
  	
  	iot
  		: topic("yo/hi")
  			: content("ping ping"): publish()
				: content("pim pom"): publish()
				: content("toc toc"): publish() 
			: topic("yo/hello")
				: content("tada"): publish()

		2: times(|index| -> iot: content("PLOP " + index): publish()) # on reste sur le topic courant

		iot: topic("yo/salut"): content("coucou"): publish()

  })
  : onSet(|res| { # if success
			println(res: getValue())
			
			iot: disconnect()
  })
  : onFail(|err| { # if failed

			println("reason " + err: getReasonCode())
			println("msg " + err: getMessage())
			println("loc " + err: getLocalizedMessage())
			println("cause " + err: getCause())
			println("excep "+ err)
			err: printStackTrace()
  })

  env: shutdown()

}

dsl.mqtt.subscribe.golo.md

module iot_four


import littleIOT


function main = |args| {

	let mybroker = broker()
		: protocol("tcp")
		: host("test.mosquitto.org")
		: port(1883)

	let options = littleIOT(): getConnectOptions()

	let iot = littleIOT(): broker(mybroker): connectOptions(options): initialize(): connect()

	iot: subscribe("/GPS/test/TESTMACHINE/#")

	iot
		: messageArrived(|topic, message| -> println("you've got a mail : " + topic + " | " + message) )
		: deliveryComplete(|token| -> println("delivery is complete " + token) )
		: connectionLost(|error| -> println("ouch"))
		: activateCallBacks()

}

mqtt.golo.md

module mqttStressTool

import org.eclipse.paho.client.mqttv3.MqttClient
import org.eclipse.paho.client.mqttv3.MqttConnectOptions
import org.eclipse.paho.client.mqttv3.MqttException
import org.eclipse.paho.client.mqttv3.MqttMessage
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence

import gololang.Async
import gololang.concurrent.workers.WorkerEnvironment

function main = |args| {

	let env = WorkerEnvironment.builder(): withCachedThreadPool()

	#...
	let topic 			= "MQTT"
	let content			= "Message from Bob"
	let qos 				= 0
	let broker			= "tcp://localhost:1883"
	let clientId		= "Bob"
	let persistence = MemoryPersistence()

  # define promise
  let talkToMQTTBroker = {
    
    return promise(): initialize(|resolve, reject| {
      env: spawn(|data| {

				try {
					let sampleClient = MqttClient(broker, clientId, persistence) # MqttClient
					let connOpts = MqttConnectOptions() # MqttConnectOptions
					connOpts: setCleanSession(true)
					println("Connecting to broker: " + broker)
					sampleClient: connect(connOpts)
					println("Connected")
					println("Publishing message: " + content)

					let message = MqttMessage(content: getBytes()) # MqttMessage
					message: setQos(qos)

					sampleClient: publish(topic, message)
					
					resolve(sampleClient)				
			    
				} catch (err) { # MqttException
						reject(err)
				}

        

      }): send("go")
    })
  }


  # run promise
  talkToMQTTBroker()
  : onSet(|sampleClient| { # if success
			println("Message published")
			sampleClient: disconnect()
			println("Disconnected")
    	java.lang.System.exit(0)
  })
  : onFail(|err| { # if failed
			println("reason " + err: getReasonCode())
			println("msg " + err: getMessage())
			println("loc " + err: getLocalizedMessage())
			println("cause " + err: getCause())
			println("excep "+ err)
			err: printStackTrace()
  })

  env: shutdown()

}