dtux__pws2mqtt/mqtt_functions.php

<?php

require "utils.php";
$mids = array();

$client = new Mosquitto\Client();

// log levels
define( "DEBUG", $client::LOG_DEBUG);		// => 16
define( "INFO", $client::LOG_INFO);			// => 1
define( "NOTICE", $client::LOG_NOTICE);		// => 2
define( "WARNING", $client::LOG_WARNING);	// => 4
define( "ERROR", $client::LOG_ERR);			// => 8
define( "ALERT", 32);
define( "ALL", DEBUG | INFO | NOTICE | WARNING | ERROR | ALERT);
$logLevel = INFO | WARNING | ERROR | ALERT;
//$notificationLevel = WARNING | ERROR;     // TODO send notification

logger(DEBUG, _("defining callback functions"));

// defining callback functions
$client->onConnect('connectResponse');
$client->onDisconnect('disconnectResponse');
$client->onSubscribe('subscribeResponse');
$client->onUnsubscribe('unsubscribeResponse');
$client->onMessage('messageReceived');
$client->onLog('logger');
$client->onPublish('publishResponse');

logger(DEBUG, _("connecting to mqtt server"));
$client->connect($mqttServerAddress, $mqttServerPort, 30);
logger(DEBUG, _("subscribing"));
$mid = $client->subscribe($topicName . "/#", 2);

function messageReceived($message)
{
	global $topicName, $logFh;
	$topic = explode ("/", $message->topic);
	if($topic[array_key_last($topic)] != "get" and ($topic[array_key_last($topic)]) != "set")
	{
		$topic = explode ("/", $message->topic, 2);   	// get topic name
		$fnTree = explode ("/" , $topic[1]); 			// get friendlyname
		echo $topic[0] . " => " . $topic[1] . EOL;
		logger(INFO, print_r(json_decode($message->payload), true));
	}
}

// payload is an array :
// $key is property => $value is value of the parameter

function publish($payload, $commande="")
{
	global $mids, $friendlyName, $topicName, $client, $logFh;
	//print_r($payload);
	$string = $topicName . "/" . $friendlyName;
	$mid = $client->publish($string, $payload , 2);
	if (isset($mids[$mid]))
	{
		//echo "unsetting mids" .EOL;
		unset ($mids[$mid]);
	}else
	{
		//echo "setting mids" .EOL;
		$mids[$mid] = true;
	}
	logger(LOG_INFO, $logFh, "Publishing " . $string . " with payload => " . $payload);
}

function connectResponse($r, $message)
{
	global $connected;
	echo sprintf(_("I got code %d and message : '%s'"), $r, $message) . EOL;
	switch ($r)
	{
		case 0:
			logger(INFO, _("Successfull connection"));
			$connected = true;
			break;
		case 1:
			logger(INFO, _("Connection refused : unacceptable protocol version"));
			$connected = false;
			break;
		case 2:
			logger(INFO, _("Connection refused : identifier rejected"));
			$connected = false;
			break;
		case 3:
			logger(INFO, _("Connection refused (broker unavailable )"));
			$connected = false;
			break;
	}
	return $connected;
}

function subscribeResponse($mid, $qosCount)
{
	global $topics;
	logger(INFO, _("Subscribed"));
}

function unsubscribeResponse($mid)
{
	global $client;
	logger(INFO, _("Unsubscribed"));
	$client->disconnect();
}

function disconnectResponse($r)
{
	global $connected;
	if ($r != 0)
	{
		$str = _('Badly ');
	}else
	{
		$str =  _('Cleanly ');
	}
	logger(INFO, $str . _("disconnected from server"));
	$connected = false;
}

function publishResponse($mid)
{
	global $mids, $events;
	logger(LOG_INFO, "Event with mid = " . $mid . " published by MQTT broker");
	if (isset($mids[$mid]))
	{
		//echo "unsetting mids" . EOL;
		unset ($mids[$mid]);
		//print_r($mids);
	}else
	{
		//echo "setting mids" . EOL;
		$mids[$mid] = true;
	}
}
?>