synchronous/blocking mode is not really supported

[Yun Wu]

It seems that blocking mode is not really supported.

Topology: publisher -- broker -- subscriber
Publisher simply publish a message then exit.
Subscriber sometimes can receive message published by publisher, but sometimes not.

The root cause is, libmosquitto does not really support blocking mode.
mosquitto_connect_bind --> _mosquitto_reconnect --> _mosquitto_send_connect
That means mosquitto_connect_bind() does not wait until CONNACK arrives.

Logs can prove that:
----
1387273535: New connection from 172.18.111.243 on port 1883.
1387273535: New client connected from 172.18.111.243 as TestClientId (c1, k3600).
1387273535: Sending CONNACK to TestClientId (0)
1387273535: Socket error on client TestClientId, disconnecting.
----
We can see:
1) Broker has not received PUBLISH message from publisher.
2) Broker has not received DISCONNECT message from publisher.

All because blocking mode is not really supported.

In blocking mode,
mosquitto_connect_bind should wait until CONNACK arrives;
If QoS not 0, mosquitto_publish should wait unit ...
mosquitto_subscribe should wait unit ...
...

Here is code of the publisher:

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "mosquitto.h"

#define dbg(fmt, args...) \
    printf("%s(%u): "fmt"\n", __FUNCTION__, __LINE__, ##args)

int main(int argc, char *argv[])
{
    struct mosquitto *mosq = NULL;
    int rv;
    const char topic[] = "TestTopic";
    const char buf[] = "Hello, world";

    if (MOSQ_ERR_SUCCESS != mosquitto_lib_init()) {
        return -1;
    }

    if (NULL == (mosq = mosquitto_new("TestClientId", 1, NULL))) {
        dbg("mosquitto_new() failed");
        rv = -1;
        goto quit;
    }

    rv = mosquitto_connect_bind(mosq, "172.18.111.245", 1883, 3600, "172.18.111.243");
    if (MOSQ_ERR_SUCCESS != rv) {
        dbg("mosquitto_connect_bind() failed: %s", mosquitto_strerror(rv));
        if (MOSQ_ERR_ERRNO == rv) {
            dbg("errno = %d: %s", errno, strerror(errno));
        }
        rv = -1;
        goto quit;
    }

    if (MOSQ_ERR_SUCCESS != mosquitto_publish(mosq, NULL, topic, sizeof(buf), buf, 0, 0)) {
        dbg("mosquitto_publish to %s failed", topic);
        rv = -1;
        goto quit;
    }

quit:
    if (NULL != mosq) {
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
    }
    mosquitto_lib_cleanup();
    return rv;
}

[Roger Light (roger.light)]

You are correct, the design of the library is intentionally asynchronous.

The only consideration of blocking is in the network socket interface itself, i.e. does the call to connect() block. This is blocking at the socket level not the protocol level.

[Roger Light (roger.light)]

The documentation says:

 * Important note
 *
 * The following functions that deal with network operations will return
 * MOSQ_ERR_SUCCESS on success, but this does not mean that the operation has
 * taken place. An attempt will be made to write the network data, but if the
 * socket is not available for writing at that time then the packet will not be
 * sent. To ensure the packet is sent, call mosquitto_loop() (which must also
 * be called to process incoming network data).
 * This is especially important when disconnecting a client that has a will. If
 * the broker does not receive the DISCONNECT command, it will assume that the
 * client has disconnected unexpectedly and send the will.
 *
 * mosquitto_connect()
 * mosquitto_disconnect()
 * mosquitto_subscribe()
 * mosquitto_unsubscribe()
 * mosquitto_publish()

You haven't called one of the mosquitto_loop*() functions, so you will
never receive a CONNACK. You've also not called mosquitto_lib_init().

If you want the publish to happen after a successful connection, put
it in the on_connect() callback. If you want to disconnect after a
successful publish, but that in the on_publish() callback and use a
while loop to call mosquitto_loop() (or just use
mosquitto_loop_forever()).

On Tue, Dec 17, 2013 at 11:05 AM, Yun Wu <[e-mail]> wrote:
> ** Description changed:
>
> It seems that blocking mode is not really supported.
>
> Topology: publisher -- broker -- subscriber
> Publisher simply publish a message then exit.
> Subscriber sometimes can receive message published by publisher, but sometimes not.
>
> The root cause is, libmosquitto does not really support blocking mode.
> mosquitto_connect_bind --> _mosquitto_reconnect --> _mosquitto_send_connect
> That means mosquitto_connect_bind() does not wait until CONNACK arrives.
>
> Logs can prove that:
> ----
> 1387273535: New connection from 172.18.111.243 on port 1883.
> 1387273535: New client connected from 172.18.111.243 as TestClientId (c1, k3600).
> 1387273535: Sending CONNACK to TestClientId (0)
> 1387273535: Socket error on client TestClientId, disconnecting.
> ----
> We can see:
> 1) Broker has not received PUBLISH message from publisher.
> 2) Broker has not received DISCONNECT message from publisher.
>
> All because blocking mode is not really supported.
>
> - In blocking mode,
> + In blocking mode,
> mosquitto_connect_bind should wait until CONNACK arrives;
> If QoS not 0, mosquitto_publish should wait unit ...
> mosquitto_subscribe should wait unit ...
> ...
> +
> +
> +
> + Here is code of the publisher:
> +
> + #include <errno.h>
> + #include <stdio.h>
> + #include <stdlib.h>
> + #include <string.h>
> + #include <unistd.h>
> + #include "mosquitto.h"
> +
> + #define dbg(fmt, args...) \
> + printf("%s(%u): "fmt"\n", __FUNCTION__, __LINE__, ##args)
> +
> + int main(int argc, char *argv[])
> + {
> + struct mosquitto *mosq = NULL;
> + int rv;
> + const char topic[] = "TestTopic";
> + const char buf[] = "Hello, world";
> +
> + if (MOSQ_ERR_SUCCESS != mosquitto_lib_init()) {
> + return -1;
> + }
> +
> + if (NULL == (mosq = mosquitto_new("TestCli...

[Yun Wu (wuyun1984-1984)]

Thanks.

mosquitto_lib_init() is already called in my codes.

In fact I just want reliable APIs for publish/subscribe/..., which means when they return SUCCESS, everything is indeed successfully finished.

Do you have any idea of implementing them based on libmosuiqtto APIs?

[Roger Light (roger.light)]

Something like this should work, assuming you really are waiting for
the publish to complete before sending another publish of course.

void on_publish(struct mosquitto *mosq, void *userdata, int mid)
{
    int *r = userdata;

    if(r){
        *r = 0;
    }
}

...
mosquitto_userdata_set(mosq, &run);
...

run = 1;
mosquitto_publish(mosq, ...);

while(run){
    rc = mosquitto_loop(mosq, -1, 1);
    if(rc != MOSQ_ERR_SUCCESS){
        /* Handle error */
    }
}

On Wed, Dec 18, 2013 at 6:40 AM, Yun Wu <[e-mail]> wrote:
> Thanks.
>
> mosquitto_lib_init() is already called in my codes.
>
> In fact I just want reliable APIs for publish/subscribe/..., which means
> when they return SUCCESS, everything is indeed successfully finished.
>
> Do you have any idea of implementing them based on libmosuiqtto APIs?
>
> --
> You received this bug notification because you are subscribed to
> mosquitto.
> https://bugs.launchpad.net/bugs/1261683
>
> Title:
> Blocking mode is not really supported
>
> To manage notifications about this bug go to:
> https://bugs.launchpad.net/mosquitto/+bug/1261683/+subscriptions

[Yun Wu (wuyun1984-1984)]

Thanks Roger.
But I just think it is still not 100% reliable because variable 'run' can be modified when publishing a earlier message, right?

[Roger Light (roger.light)]

Ah, well now you've changed what you're asking :)

On Thu, Dec 19, 2013 at 10:56 AM, Yun Wu <[e-mail]> wrote:
> Thanks Roger.
> But I just think it is still not 100% reliable because variable 'run' can be modified when publishing a earlier message, right?
>
> --
> You received this bug notification because you are subscribed to
> mosquitto.
> https://bugs.launchpad.net/bugs/1261683
>
> Title:
> synchronous/blocking mode is not really supported
>
> To manage notifications about this bug go to:
> https://bugs.launchpad.net/mosquitto/+bug/1261683/+subscriptions

[Yun Wu (wuyun1984-1984)]

Not yet, still about reliable synchronous API :-)

[Roger Light (roger.light)]

My point is that if you set run=1 before each call to publish(), then wait for that publish to complete before doing anything else (which is what you want), then it will work fine (assuming the broker doesn't disconnect etc).

If an earlier call to publish() can have an effect on the run variable for a subsequent publish, then you obviously didn't use a blocking publish in the first case.

Another way to achieve a similar effect would be to store the mid of the blocking message, then compare it to the highest mid sent out (by using the on_publish callback). If my mid is 1 and the highest sent mid is 2, then my message has sent. You'd have to deal with mid wraparound as well of course.

[Yun Wu (wuyun1984-1984)]

The 1st method, if a process has 2 threads, one use synchronous publish and another asynchronous publish, then what I worried may happen, right?

I think the 2nd method should work.

[Roger Light (roger.light)]

Yes you are correct, if you try to mix synchronous and asynchronous
publishing then you will have problems with the first method. The
second method should work fine in this situation though.

On Mon, Dec 23, 2013 at 8:55 AM, Yun Wu <[e-mail]> wrote:
> The 1st method, if a process has 2 threads, one use synchronous publish
> and another asynchronous publish, then what I worried may happen,
> right?
>
> I think the 2nd method should work.
>
> --
> You received this bug notification because you are subscribed to
> mosquitto.
> https://bugs.launchpad.net/bugs/1261683
>
> Title:
> synchronous/blocking mode is not really supported
>
> To manage notifications about this bug go to:
> https://bugs.launchpad.net/mosquitto/+bug/1261683/+subscriptions

[Yun Wu (wuyun1984-1984)]

OK. Thanks very much.