Marketplace:PHP access AMQP

Background overview

The third-party PHP library does not have a solution that supports AMQP 1.0. In order to facilitate PHP developers to access AMQP, we provide extended support for STOMP.

For pre-configuration, please refer to the parent document. This article only introduces the part of running the client.

Run Client

Download the SDK

This example provides an example based on Stomp PHP the code example for the library, using STOMP protocol and IoT platform cloud communication. Please visit Stomp PHP download the client and view usage instructions.

Stomp PHP SDK applicable PHP version, see composer.json in the Stomp PHP SDK medium require statement.

Because Stomp PHP 5.0.0 the following versions exist SDK the problem may not be reconnected after disconnection. It is recommended that you download it. Stomp PHP 5.0.0 or its Above version the SDK. For detailed instructions, see Issues .

You can run the following command in the PHP project directory to download the SDK of Stomp PHP version 5.0.0.

composer require stomp-php/stomp-php 5.0.0

Code example

STOMP Client

<?php
  require __DIR__ . '/vendor/autoload.php';
use Stomp\Client;
use Stomp\Network\Observer\Exception\HeartbeatException;
use Stomp\Network\Observer\ServerAliveObserver;
use Stomp\StatefulStomp;
use Stomp\Transport\Frame;

//For parameter descriptions, see the AMQP Client Access Instructions document.
// Project code leakage may lead to AccessKey leakage and threaten the security of all resources under the account. The following code example uses environment variables to obtain AccessKey for calling, for reference only.
$accessKey = "${YourAccessKey}""; //Recommended use: getenv("ACCESS_KEY")
$accessSecret = "${YourAccessSecret}""; // Recommended use: getenv("ACCESS_SECRET") 
$consumerGroupId = "${YourConsumerGroupId}"";
$clientId = "${YourClientID}"";	//Custom
//iotInstanceId：Instance ID, can be empty。
$iotInstanceId = "";
$timeStamp = round(microtime(true) * 1000);
//Signature method: supports hmacmd5, hmacsha1 and hmacsha256.
$signMethod = "hmacsha1";
//For the userName assembly method, see the AMQP client access instructions.
//If binary transmission is used, the userName needs to add the encode=base64 parameter, and the server will encode the message body in base64 before pushing it. For specific adding methods, please refer to the next section "Binary Message Body Description".
$userName = $clientId . "|authMode=aksign"
  . ",signMethod=" . $signMethod
  . ",timestamp=" . $timeStamp
  . ",authId=" . $accessKey
  . ",iotInstanceId=" . $iotInstanceId
  . ",consumerGroupId=" . $consumerGroupId
  . "|";
$signContent = "authId=" . $accessKey . "&timestamp=" . $timeStamp;
//For signature calculation and password assembly methods, please refer to the AMQP client access instructions.
$password = base64_encode(hash_hmac("sha1", $signContent, $accessSecret, $raw_output = TRUE));
//For the access domain name, see Instance Console - Development Configuration - AMQP.
$client = new Client('tcp://${YourHost}":61613');

//Server heartbeat monitoring.
$observer = new ServerAliveObserver();
$client->getConnection()->getObservers()->addObserver($observer);
//Heartbeat setting, no need for the cloud to send heartbeat packets
$client->setHeartbeat(30000, 0);
//Set read to non-blocking mode
$client->getConnection()->setReadTimeout(0);
$client->setLogin($userName, $password);
try {
  $client->connect();
  echo "Connection successful！", PHP_EOL;
}
catch(Exception $e) {
  echo "Failed to connect to the server, error message: " . $e->getMessage() , PHP_EOL;
  exit(1);
}

// Use direct Frame subscription instead of StatefulStomp
$frame = new Frame('SUBSCRIBE');
$frame->addHeaders([
                   'destination' => $consumerGroupId, // Use the consumer group ID you configured
                   'id' => $consumerGroupId,
                   'ack' => 'auto'
                   ]);
$client->sendFrame($frame);
$currentTime = date('Y-m-d H:i:s');
echo "[{$currentTime}]Subscription Success！", PHP_EOL;

// Initialize heartbeat related variables
$lastHeartbeatTime = time();
while (true) {
  try {
    // Check the connection status
    if (!$client->isConnected()) {
      $currentTime = date('Y-m-d H:i:s');
      echo "[{$currentTime}] The connection does not exist. Reconnect after 10 seconds.", PHP_EOL;
      sleep(10);
      $client->connect();

      $frame = new Frame('SUBSCRIBE');
      $frame->addHeaders([
                         'destination' => $consumerGroupId,
                         'id' => $consumerGroupId,
                         'ack' => 'auto'
                         ]);
      $client->sendFrame($frame);

      $currentTime = date('Y-m-d H:i:s');
      echo "[{$currentTime}]Reconnect and subscribe successfully", PHP_EOL;
    }

    // Read message frames directly from the client
    $frame = $client->readFrame();
    if ($frame) {
      $currentTime = date('Y-m-d H:i:s');
      echo "[{$currentTime}] Receive message：", PHP_EOL;
      echo "Headers: ", print_r($frame->getHeaders(), true), PHP_EOL;
      echo "Body: ", $frame->getBody(), PHP_EOL;
        }
    else {
      // Send a heartbeat packet every 1 second
      $now = time();
      if ($now - $lastHeartbeatTime >= 30) {

          $heartbeatFrame = new Frame('STOMP');
          $client->sendFrame($heartbeatFrame);               
          $lastHeartbeatTime = $now;
          echo "[{$currentTime}] Sending heartbeat packets", PHP_EOL;
        } 
      }
    }
    catch(HeartbeatException $e) {
        $currentTime = date('Y-m-d H:i:s');
        echo "[{$currentTime}] The server did not send a heartbeat packet within the specified time", PHP_EOL;
        $client->disconnect();
    } catch(Exception $e) {
        $currentTime = date('Y-m-d H:i:s');
        echo "[{$currentTime}] Error processing message: ". $e->getMessage() , PHP_EOL;
        $client->disconnect();
    }
}   

The parameter description is as follows: