Transferring data over socket failed

[pheelaphong]

Sometimes find this error

---

PhpMqtt \ Client \ Exceptions \ DataTransferException (66)
[66] Transferring data over socket failed: Reading data from the socket failed. Has it been closed?

---

[Namoshek]

That's a genuine error when the connection has been lost, as the message states. It is most likely related to your code, e.g. because you are not calling $mqtt->loop(). If you don't call this method, there won't be sent any pings to the broker and the after some time, the broker will consider your client disconnected and close the socket.

If you have problems, please share some code and probably also logs. It makes it a lot easier to help.

[pheelaphong]

That's a genuine error when the connection has been lost, as the message states. It is most likely related to your code, e.g. because you are not calling $mqtt->loop(). If you don't call this method, there won't be sent any pings to the broker and the after some time, the broker will consider your client disconnected and close the socket.

If you have problems, please share some code and probably also logs. It makes it a lot easier to help.

public function mqtt_pub(Request $request)
    {
        $topic = $request->route('topic');
        $msg = $request->route('msg');
        $mqtt = MQTT::connection();
        $mqtt->publish('cmnd/' . $topic . '/POWER', $msg, 0);

        /** @var \PhpMqtt\Client\Contracts\MqttClient $mqtt */
        $result = [];

        $mqtt = MQTT::connection();
        $mqtt->subscribe('stat/' . $topic . '/POWER', function (string $topic, string $message) use ($mqtt, &$result) {
            $result['topic'] = $topic;
            $result['message'] = $message;

            $mqtt->interrupt();
        }, 1);
        $mqtt->loop(true);

        return response()->json($result, 200);
        //$this->mqtt_sub();
    }
    
    public function mqtt_sub(Request $request)
    {
        $topic = $request->route('topic');
        $mqtt = MQTT::connection();
        $mqtt->publish('cmnd/' . $topic . '/POWER', '', 0);
        $result = [];
        $mqtt->subscribe('stat/' . $topic . '/POWER', function (string $topic, string $message) use ($mqtt, &$result) {
            $result['topic'] = $topic;
            $result['message'] = $message;

            $mqtt->interrupt();
        }, 1);
        $mqtt->loop(true);
        return response()->json($result, 200);
    }

Sorry, this is my first time.

[pheelaphong]

$( document ).ready(function() {
    subscribe('xxxxx');
});
    
function subscribe(topic) {
    $.ajax({
        url: 'mqttsub/' + topic,
        dataType: 'JSON',
        cache: false,
        success: function(res) {
            console.log(res.message);
            if (res.message === "ON") {
                 //ON
            } else {
                 //OFF
            }
            setTimeout(function() {
                subscribe(topic)
            }, 1000);
        },
        statusCode: {
            500: function() {
                console.log('reconnect');
                subscribe(topic)
            }
        }
    });
}

[Namoshek]

I don't see anything wrong with the code. It isn't really clean but it should do the job. Without further infos regarding MQTT client configuration and some logs, I don't think I'm able to help.

[pheelaphong]

configuration

'default' => [

            // The host and port to which the client shall connect.
            'host' => 'xxxx',
            'port' => 1883,

            // The MQTT protocol version used for the connection.
            'protocol' => MqttClient::MQTT_3_1,

            // A specific client id to be used for the connection. If omitted,
            // a random client id will be generated for each new connection.
            //'client_id' => "ClientID".rand(),
            'client_id' => env('MQTT_CLIENT_ID'),

            // Whether a clean session shall be used and requested by the client.
            // A clean session will let the broker forget about subscriptions and
            // queued messages when the client disconnects. Also, if available,
            // data of a previous session will be deleted when connecting.
            'use_clean_session' => env('MQTT_CLEAN_SESSION', true),

            // Whether logging shall be enabled. The default logger will be used
            // with the log level as configured.
            'enable_logging' => env('MQTT_ENABLE_LOGGING', true),

            // Defines which repository implementation shall be used. Currently,
            // only a MemoryRepository is supported.
            'repository' => MemoryRepository::class,

            // Additional settings used for the connection to the broker.
            // All of these settings are entirely optional and have sane defaults.
            'connection_settings' => [

                // The TLS settings used for the connection. Must match the specified port.
                'tls' => [
                    'enabled' => env('MQTT_TLS_ENABLED', false),
                    'allow_self_signed_certificate' => env('MQTT_TLS_ALLOW_SELF_SIGNED_CERT', false),
                    'verify_peer' => env('MQTT_TLS_VERIFY_PEER', true),
                    'verify_peer_name' => env('MQTT_TLS_VERIFY_PEER_NAME', true),
                    'ca_file' => env('MQTT_TLS_CA_FILE'),
                    'ca_path' => env('MQTT_TLS_CA_PATH'),
                    'client_certificate_file' => env('MQTT_TLS_CLIENT_CERT_FILE'),
                    'client_certificate_key_file' => env('MQTT_TLS_CLIENT_CERT_KEY_FILE'),
                    'client_certificate_key_passphrase' => env('MQTT_TLS_CLIENT_CERT_KEY_PASSPHRASE'),
                ],

                // Credentials used for authentication and authorization.
                'auth' => [
                    'username' => env('MQTT_AUTH_USERNAME'),
                    'password' => env('MQTT_AUTH_PASSWORD'),
                ],

                // Can be used to declare a last will during connection. The last will
                // is published by the broker when the client disconnects abnormally
                // (e.g. in case of a disconnect).
                'last_will' => [
                    'topic' => env('MQTT_LAST_WILL_TOPIC'),
                    'message' => env('MQTT_LAST_WILL_MESSAGE'),
                    'quality_of_service' => env('MQTT_LAST_WILL_QUALITY_OF_SERVICE', 0),
                    'retain' => env('MQTT_LAST_WILL_RETAIN', false),
                ],

                // The timeouts (in seconds) used for the connection. Some of these settings
                // are only relevant when using the event loop of the MQTT client.
                'connect_timeout' => env('MQTT_CONNECT_TIMEOUT', 60),
                'socket_timeout' => env('MQTT_SOCKET_TIMEOUT', 5),
                'resend_timeout' => env('MQTT_RESEND_TIMEOUT', 10),

                // The interval (in seconds) in which the client will send a ping to the broker,
                // if no other message has been sent.
                'keep_alive_interval' => env('MQTT_KEEP_ALIVE_INTERVAL', 10),
            ],

        ]

I just changed host, username, password

this code working. but just sometime failed

[pheelaphong]

.env

---

MQTT_HOST=xxx
MQTT_AUTH_USERNAME=xxx
MQTT_AUTH_PASSWORD=xxx
MQTT_CLIENT_ID=1234567

---

[Namoshek]

Weird. Please also let me know what broker you are using (Mosquitto, HiveMQ, EMQ X, ...) and show some logs. If you configure logging to use debug as level, also the MQTT client will log with debug level. It is definitely weird behavior though.

[pheelaphong]

I edited
Now working, not error
I changed
'client_id' => env('MQTT_CLIENT_ID'),
to
'client_id' => env('MQTT_CLIENT_ID').rand(),
I don't know if it is related or not, but it working

I have more questions in frontend

You have an idea about subscribe. without loop setTimeout
Now I do loops all the time.
From my code

$( document ).ready(function() {
    subscribe('xxxxx');
});
    
function subscribe(topic) {
    $.ajax({
        url: 'mqttsub/' + topic,
        dataType: 'JSON',
        cache: false,
        success: function(res) {
            console.log(res.message);
            if (res.message === "ON") {
                 //ON
            } else {
                 //OFF
            }
            setTimeout(function() {
                subscribe(topic)
            }, 1000);
        },
        statusCode: {
            500: function() {
                console.log('reconnect');
                subscribe(topic)
            }
        }
    });
}

Hope you understand I'm not good English

[Namoshek]

Oh yeah, sure. I know what the problem is. It's actually quite obvious. You are connecting to the broker with the same client id from multiple HTTP requests. The clients "kick each other" from the server, causing disconnects.

If you do not specify MQTT_CLIENT_ID in the .env file, a random one will be generated for you. So you do not have to do this manually.

[Namoshek]

Since the cause of the problem has been discussed, I'm closing this issue. Regarding your ajax question, I cannot help. This sounds like a common question though, so you might find an answer on StackOverflow or so.

[3urdparty]

Thanks @Namoshek I've been stuck on this issue for a while. I didn't know about subtlety.

[shekharhhp]

I am using below code but getting below error
Connected to MQTT Broker
Subscribed to topic 'device/data'
Error while listening for messages: [66] Transferring data over socket failed:
Reading data from the socket failed. Has it been closed?

so please help me here

<?php

require 'vendor/autoload.php';

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;

class MqttClientHandler {

    private $mqttClient;

    public function __construct($host, $port, $clientId) {
        $this->mqttClient = new MqttClient($host, $port, $clientId);
    }

    public function connect($username, $password) {
        $settings = (new ConnectionSettings())
            ->setUsername($username)
            ->setPassword($password)
            ->setKeepAliveInterval(15)
            ->setConnectTimeout(60)
            ->setSocketTimeout(120) 
            ->setUseTls(false);

        try {
            $this->mqttClient->connect($settings, true);
            echo "Connected to MQTT Broker\n";
        } catch (Exception $e) {
            echo "Failed to connect: " . $e->getMessage() . "\n";
        }
    }

    public function sendMessage($topic, $message) {
        try {
            $this->mqttClient->publish($topic, $message, 0);
            echo "Message published to topic '$topic'\n";
        } catch (Exception $e) {
            echo "Failed to publish message: " . $e->getMessage() . "\n";
        }
    }

    public function subscribe($topic, $callback) {
        try {
            $this->mqttClient->subscribe($topic, function ($topic, $message) use ($callback) {
                $callback($topic, $message);
            }, 0);

            echo "Subscribed to topic '$topic'\n";
        } catch (Exception $e) {
            echo "Failed to subscribe: " . $e->getMessage() . "\n";
        }
    }

    public function listen() {
        try {
            while (true) {
                if (!$this->mqttClient->isConnected()) {
                    echo "Reconnecting...\n";
                    $this->mqttClient->connect();
                }
                $this->mqttClient->loop(true);
                usleep(100000);
            }
        } catch (Exception $e) {
            echo "Error while listening for messages: " . $e->getMessage() . "\n";
        }
    }    

    public function disconnect() {
        $this->mqttClient->disconnect();
        echo "Disconnected from MQTT Broker\n";
    }
} and <?php

require 'mqtt_client.php'; // Include the class

// MQTT Broker Details
define('MQTT_HOST', 'xxxxx');
define('MQTT_PORT', 1883);
define('MQTT_USERNAME', 'xxxxx');
define('MQTT_PASSWORD', 'xxxxx');
define('MQTT_CLIENT_ID', 'xxxxx');

// Create a new MQTT client handler
$mqttHandler = new MqttClientHandler(MQTT_HOST, MQTT_PORT, MQTT_CLIENT_ID);

// Callback function to handle incoming messages
function onMessageReceived($topic, $message) {
    // Process incoming message
    echo "Received message: '$message' on topic: '$topic'\n";
    
    // You can perform any action here like saving to a database or triggering an event
    // For example, save the message to a file
    file_put_contents('received_data.txt', "Topic: $topic - Message: $message\n", FILE_APPEND);
}

// Connect to the MQTT Broker
$mqttHandler->connect(MQTT_USERNAME, MQTT_PASSWORD);

// Subscribe to the topic (adjust topic as per your needs)
$topic = 'device/data'; // Example topic
$mqttHandler->subscribe($topic, 'onMessageReceived');

// Start listening for incoming messages (blocking call)
$mqttHandler->listen();
?>

[Namoshek]

@shekharhhp Please do remove credentials from your code before posting it. I've removed them for now and also removed the post revision to hide it, but to be safe, you might want to change your MQTT password on the server.

Regarding your question, please open a new issue with the formatted code. What would also be helpful to diagnose the issue is timestamps in the logged messages, e.g. a microtime(true) at the start of each line.

[shekharhhp]

I tried but still getting same errors here <?php

require 'vendor/autoload.php';
use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;

class MqttClientHandler {

private $mqttClient;

public function __construct($host, $port, $clientId) {
    $this->mqttClient = new MqttClient($host, $port, $clientId);
}

public function log($message) {
    echo "[" . date('Y-m-d H:i:s') . "] $message\n";
}

public function connect($username, $password) {
    $settings = (new ConnectionSettings())
        ->setUsername($username)
        ->setPassword($password)
        ->setKeepAliveInterval(30)
        ->setConnectTimeout(60)
        ->setSocketTimeout(60)
        ->setUseTls(false);

    try {
        $this->mqttClient->connect($settings, true);
        $this->log("Connected to MQTT Broker");
    } catch (Exception $e) {
        $this->log("Failed to connect: " . $e->getMessage());
    }
}

public function sendMessage($topic, $message) {
    try {
        $this->mqttClient->publish($topic, $message, 0);
        $this->log("Message published to topic '$topic'");
    } catch (Exception $e) {
        $this->log("Failed to publish message: " . $e->getMessage());
    }
}

public function subscribe($topic, $callback) {
    try {
        $this->mqttClient->subscribe($topic, function ($topic, $message) use ($callback) {
            $callback($topic, $message);
        }, 0);
        $this->log("Subscribed to topic '$topic'");
    } catch (Exception $e) {
        $this->log("Failed to subscribe: " . $e->getMessage());
    }
}

public function listen() {
    try {
        while (true) {
            if (!$this->mqttClient->isConnected()) {
                $this->log("Reconnecting...");
                $this->connect(MQTT_USERNAME, MQTT_PASSWORD);
            }
            $this->mqttClient->loop(true);
            usleep(10000);  // Short delay for responsiveness
        }
    } catch (Exception $e) {
        $this->log("Error while listening for messages: " . $e->getMessage());
    }
}

public function disconnect() {
    $this->mqttClient->disconnect();
    $this->log("Disconnected from MQTT Broker");
}

}