mqtt/MQTT.php
public function publish_sync($topic, $message, $qos=0, $retain=0, &$msgid=0)
    {
        # set default call_handler

        # initial dup = 0
        $dup = 0;
        # initial msgid = 0
        $msgid = 0;

        # non blocking
        $this->socket->set_non_blocking();

        $r = $this->do_publish($topic, $message, $qos, $retain, $msgid, $dup);

        if ($qos == 0) {
            return $r['ret'];
        }

        # loop
        do {
            $r = $this->handle_message();
            if (!$r) {
                usleep(10000);
                continue;
            }

            $finished = $qos == 1 ?
                $this->cmdstore->isEmpty(Message::PUBACK, $msgid) :
                (
                    $this->cmdstore->isEmpty(Message::PUBREC, $msgid) &&
                    $this->cmdstore->isEmpty(Message::PUBCOMP, $msgid)
                );

            if (!$finished) {
                # retry publish
                $this->handle_publish($msgid);
            } else {
                return true;
            }

        } while (true);

        return false;
    }
public function publish_sync($topic, $message, $qos=0, $retain=0, &$msgid=0)
    {
        # set default call_handler

        # initial dup = 0
        $dup = 0;
        # initial msgid = 0
        $msgid = 0;

        # non blocking
        $this->socket->set_non_blocking();

        $r = $this->do_publish($topic, $message, $qos, $retain, $msgid, $dup);

        if ($qos == 0) {
            return $r['ret'];
        }

        # loop
        do {
            $r = $this->handle_message();
            if (!$r) {
                usleep(10000);
                continue;
            }

            $finished = $qos == 1 ?
                $this->cmdstore->isEmpty(Message::PUBACK, $msgid) :
                (
                    $this->cmdstore->isEmpty(Message::PUBREC, $msgid) &&
                    $this->cmdstore->isEmpty(Message::PUBCOMP, $msgid)
                );

            if (!$finished) {
                # retry publish
                $this->handle_publish($msgid);
            } else {
                return true;
            }

        } while (true);

        return false;
    }
public function publish_sync ( $ topic , $ message , $ qos = 0 , $ retain = 0 , & $ msgid = 0 ) { # set default call_handler # initial dup = 0 $ dup = 0 ; # initial msgid = 0 $ msgid = 0 ; # non blocking $ this -> socket -> set_non_blocking ( ) ; $ r = $ this -> do_publish ( $ topic , $ message , $ qos , $ retain , $ msgid , $ dup ) ; if ( $ qos == 0 ) { return $ r [ 'ret' ] ; } # loop do { $ r = $ this -> handle_message ( ) ; if ( ! $ r ) { usleep ( 10000 ) ; continue ; } $ finished = $ qos == 1 ? $ this -> cmdstore -> isEmpty ( Message :: PUBACK , $ msgid ) : ( $ this -> cmdstore -> isEmpty ( Message :: PUBREC , $ msgid ) && $ this -> cmdstore -> isEmpty ( Message :: PUBCOMP , $ msgid ) ) ; if ( ! $ finished ) { # retry publish $ this -> handle_publish ( $ msgid ) ; } else { return true ; } } while ( true ) ; return false ; }
Publish Message to topic synchronized

@param string   $topic
@param string   $message
@param int      $qos
@param int      $retain
@param int    & $msgid
@return array|bool
@throws Exception
Publish Message to topic synchronized