Issue with Publish Function Blocking Code During Callback Execution

[vorayash]

When I perform an operation in a callback that takes some time, the publish function blocks the code. Ideally, it should return a token, but it's blocking the code.

Example Code:

const int QOS = 1;
const int N_RETRY_ATTEMPTS = 5;


void log(const char* log)
{
	auto now = std::chrono::system_clock::now();
	std::time_t now_c = std::chrono::system_clock::to_time_t(now);

	std::cout << std::put_time(std::localtime(&now_c), "%Y-%m-%d %H:%M:%S")
		<< " " << log << std::endl;
}

void log(std::string log)
{
	auto now = std::chrono::system_clock::now();
	std::time_t now_c = std::chrono::system_clock::to_time_t(now);

	std::cout << std::put_time(std::localtime(&now_c), "%Y-%m-%d %H:%M:%S")
		<< " " << log << std::endl;
}


class callback : public virtual mqtt::callback,
	public virtual mqtt::iaction_listener
{
	mqtt::async_client& cli_;
	mqtt::connect_options& connOpts_;
	int nretry_;

	void reconnect() {
		std::this_thread::sleep_for(std::chrono::milliseconds(2500));
		try {
			cli_.connect(connOpts_, nullptr, *this);
		}
		catch (const mqtt::exception& exc) {
			log(std::string("Error: ") + exc.what());
		}
	}

	// Re-connection failure handler
	void on_failure(const mqtt::token& tok) override {
		log("Connection attempt failed");
		if (++nretry_ > N_RETRY_ATTEMPTS) return;
		reconnect();
	}

	// Successful connection handler
	void on_success(const mqtt::token& tok) override {}

	// Callback when connection lost
	void connection_lost(const std::string& cause) override {
		log("Connection lost");
		if (!cause.empty())
			log("Cause: " + cause);
		log("Reconnecting...");
		nretry_ = 0;
		reconnect();
	}

	// Callback when a message arrives
	void message_arrived(mqtt::const_message_ptr msg) override {
		log("Message arrived on topic: " + msg->get_topic() + "\n"
			+ "Payload: " + msg->to_string());
	}

	// Delivery complete callback
	void delivery_complete(mqtt::delivery_token_ptr token) override {}

	// New connected callback
	void connected(const std::string& cause) override {
		log("Successfully connected to MQTT server!");
		std::this_thread::sleep_for(std::chrono::milliseconds(10000)); // 0.5 sec

		if (!cause.empty()) {
			log("Connection cause: " + cause);
		}
	}

public:
	callback(mqtt::async_client& cli, mqtt::connect_options& connOpts)
		: cli_(cli), connOpts_(connOpts), nretry_(0) {}
};

int main(int argc, char* argv[])
{
	mqtt::async_client client(SERVER_ADDRESS, CLIENT_ID);

	// TLS/SSL Options
	mqtt::ssl_options sslopts;
	sslopts.set_trust_store("ras.root.ca.crt");        // Root CA certificate
	sslopts.set_key_store("device_certificate.pem");      // Client certificate
	sslopts.set_private_key("device_private_key.pem");    // Client private key
	sslopts.set_enable_server_cert_auth(true);         // Verify server cert
	// Optional: sslopts.set_alpn_protos({"mqtt"});   // if broker requires ALPN

	mqtt::connect_options connOpts;
	connOpts.set_clean_session(true);
	connOpts.set_ssl(sslopts);

	callback cb(client, connOpts);
	client.set_callback(cb);

	try {
		log("Connecting to the MQTT server...");
		client.connect(connOpts, nullptr, cb)->wait_for(2000);
		log("Connected!");

		auto msg = mqtt::make_message(TOPIC1, "Hello from Paho MQTT C++!");
		msg->set_qos(QOS);

		log("Trying to publish.");
		client.publish(msg);
		log("Message published.");

		// Keep the client alive to receive messages
		std::this_thread::sleep_for(std::chrono::seconds(10));

		log("Disconnecting...");
		client.disconnect()->wait();
		log("Disconnected.");
	}
	catch (const mqtt::exception& exc) {

		log(std::string("error") + exc.what());
		return 1;
	}

	return 0;
}

Output:

2025-11-26 23:56:40 Connecting to the MQTT server...
2025-11-26 23:56:41 Successfully connected to MQTT server!
2025-11-26 23:56:41 Connected!
2025-11-26 23:56:41 Trying to publish.
2025-11-26 23:56:51 Connection cause: connect onSuccess called
2025-11-26 23:56:51 Message published.
2025-11-26 23:57:01 Disconnecting...
2025-11-26 23:57:01 Disconnected.

As you can see in oupput, publish is finished after callback completed eventhough I am not waiting on publish token.

[fpagliughi]

Yes. Callbacks run in the context of the library thread which processes packets. You should not block it for long amounts of time, for any reason, and certainly not to wait for an operation to complete; that would cause a deadlock. If you need to do any long operations, you should spawn them off to a separate thread.

[vorayash]

@fpagliughi I’m using the async client here, but it seems that the publish function is waiting for the callback to complete. Shouldn't it just return the token and continue execution? It should only block if I explicitly call token->wait(), correct? Could you help clarify why this is happening?