composer require jesusslim/mqttclient

"minimum-stability": "dev"

swoole 2.0.8+(swoole.so)
mosquitto.so

class Logger implements \mqttclient\src\swoole\MqttLogInterface {

	public function log($type,$content,$params = []){
	        echo "$type : $content \r\n";
	 }
}

$host = '127.0.0.1';
$port = 1883;
$r = new \mqttclient\src\mosquitto\MqttClient($host,$port,10017);
$r->setAuth('username','password');
$r->setKeepAlive(60);
$r->setLogger(new Logger());
$r->setMaxReconnectTimesWhenError(360*12);
//reconnect interval
$r->setReconnectInterval(10);
//subscribe topics,callback's params can be any data we mapped into the container(IOC)
$r->setTopics(
[
    new \mqttclient\src\subscribe\Topic('test/slim',function($msg){
        echo "I receive:".$msg."\r\n";}),
    new \mqttclient\src\subscribe\Topic('test/slim3',function(\mqttclient\src\swoole\MqttClient $client,$msg){
        echo "I receive:".$msg." for slim3 \r\n";
        echo $client->getClientId();
    })
]
);
//set trigger
$r->on(\mqttclient\src\consts\ClientTriggers::SOCKET_CONNECT,function(){
    //do something
});
$r->start();

$host = '127.0.0.1';
$port = 1883;
$r = new \mqttclient\src\mosquitto\MqttSender($host,$port,10017);
$r->setAuth('username','password');
$r->setKeepAlive(60);
$r->setLogger(new Logger());
$r->setMaxReconnectTimesWhenError(360*12);
//reconnect interval
$r->setReconnectInterval(10);
$r->setQueue(new Queue());
$r->start();

class Logger implements \mqttclient\src\swoole\MqttLogInterface {

	public function log($type,$content,$params = []){
	        echo "$type : $content \r\n";
	 }
}

class Store implements \mqttclient\src\swoole\TmpStorageInterface{

	private $data = [];

    public function set($message_type, $key, $sub_key, $data, $expire = 3600)
    {
        $this->data[$message_type][$key][$sub_key] = $data;
    }

    public function get($message_type, $key, $sub_key)
    {
        return $this->data[$message_type][$key][$sub_key];
    }

    public function delete($message_type, $key, $sub_key)
    {
        if (!isset($this->data[$message_type][$key][$sub_key])){
            echo "storage not found:$message_type $key $sub_key";
        }
        unset($this->data[$message_type][$key][$sub_key]);
    }

}

$host = '127.0.0.1';
$port = 1883;

$r = new \mqttclient\src\swoole\MqttClient($host,$port,10017);
$r->setAuth('username','password');
$r->setKeepAlive(60);
$r->setLogger(new Logger());
$r->setStore(new Store());
//dns lookup
$r->setDnsLookup(true);
//buffer size
$r->setSocketBufferSize(1024*1024*5);
//reconnect times when error
$r->setMaxReconnectTimesWhenError(360*12);
//reconnect interval
$r->setReconnectInterval(10000);
//subscribe topics,callback's params can be any data we mapped into the container(IOC)
$r->setTopics(
[
    new \mqttclient\src\subscribe\Topic('test/slim',function($msg){
        echo "I receive:".$msg."\r\n";}),
    new \mqttclient\src\subscribe\Topic('test/slim3',function(\mqttclient\src\swoole\MqttClient $client,$msg){
        echo "I receive:".$msg." for slim3 \r\n";
        echo $client->getClientId();
    })
]
);

//set trigger
$r->on(\mqttclient\src\consts\ClientTriggers::RECEIVE_SUBACK,function(\mqttclient\src\swoole\MqttClient $client){
	$client->publish('slim/echo','GGXX',\mqttclient\src\consts\Qos::ONE_TIME);
});

$r->connect();
$r->publish('test/slim','test qos',2);

class Client extends MqttClient
{
    private $mysql_handler;
    private $mongo_handler;

    public function __construct($host,$port,$client_id,$mysql_conf,$mongo_conf)
    {
        $this->mysql_handler = new Mysqli($mysql_conf);
        $this->mongo_handler = new \MongoClient('mongodb://'.$mongo_conf['username'].':'.$mongo_conf['password'].'@'.$mongo_conf['host'].':'.$mongo_conf['port'].'/'.$mongo_conf['db']);
        parent::__construct($host,$port,$client_id);
    }

	 /**
     * override the produceContainer function and map your own class/data/closure to the injector,and they can be used in every publish receive handler
     * for exp: $client->setTopics([new Topic('test/own',function($mongo,$msg){ $result = $mongo->selectCollection('log_platform','test')->find(['sid' => ['$gte' => intval($msg)]]); })]);
     * @return Injector
     */
    protected function produceContainer()
    {
        $container = new Injector();
        $container->mapData(MqttClient::class,$this);
        $container->mapData(Client::class,$this);
        $container->mapData('mysqli',$this->mysql_handler);
        $container->mapData('mongo',$this->mongo_handler);
        return $container;
    }

}