When the pubsub pattern is used to transmit asynchronous APIs via UDP, messages may sometimes fail to be received. The specific behavior is as follows: the message cannot be received at the current time, but when another message is sent next time, the message sent last time can be received.
Using the following demo, when I send 3 messages from the pub side at a relatively fast speed, the sub side can sometimes only receive 1 message.
Could you please help analyze whether this issue is caused by my improper usage or a problem with the protocol stack? Thank you very much.
// pub.c
`
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <nng/nng.h>
#define SERVER_URL "udp://192.168.31.3:43017"
#define MESSAGE_COUNT 10
int main() {
nng_socket sock;
int ret;
nng_init(NULL);
if ((ret = nng_pub0_open(&sock)) != 0) {
fprintf(stderr, "Failed to open pub socket: %s\n", nng_strerror(ret));
return 1;
}
if ((ret = nng_listen(sock, SERVER_URL, NULL, 0)) != 0) {
fprintf(stderr, "Failed to listen: %s\n", nng_strerror(ret));
nng_socket_close(sock);
return 1;
}
printf("Publisher started on %s\n", SERVER_URL);
printf("Publishing %d messages...\n", MESSAGE_COUNT);
// Ten messages were sent here, but the sub only received one due to its slow reception speed.
for (int i = 0; i < MESSAGE_COUNT; i++) {
char msg[128];
snprintf(msg, sizeof(msg), "Message %d from publisher", i);
if ((ret = nng_send(sock, msg, strlen(msg), 0)) != 0) {
fprintf(stderr, "Failed to send message: %s\n", nng_strerror(ret));
break;
}
printf("Sent: %s\n", msg);
}
nng_socket_close(sock);
printf("Publisher finished\n");
printf("wait to finish...\n");
while (1) {
sleep(1);
}
return 0;
}
`
// sub.c
`
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <nng/nng.h>
void recv_cb(void *arg);
typedef struct {
nng_socket sock;
nng_aio *aio_recv;
} node_t;
node_t node;
// The sub has only received one message.
void recv_cb(void *arg) {
printf("app recv msg !!!\n");
if (nng_aio_result(node.aio_recv) != 0) {
fprintf(stderr, "Receive error: %s\n", nng_strerror(nng_aio_result(node.aio_recv)));
return;
}
nng_msg *msg = nng_aio_get_msg(node.aio_recv);
int len = nng_msg_len(msg);
char buf[1024] = {0};
memcpy(buf, (char*)nng_msg_body(msg), len);
printf("PUBSUB Aio RecvMsg:");
for (int i = 0; i < len; i++) {
printf("%x ", buf[i]);
}
printf("\n");
nng_msg_free(msg);
sleep(5);
nng_socket_recv(node.sock, node.aio_recv);
}
int main(int argc, char **argv) {
if (argc < 2) {
fprintf(stderr, "%s \n", argv[0]);
return 0;
}
int rv;
nng_init(NULL);
// init socket
if ((rv = nng_sub0_open(&node.sock)) != 0) {
fprintf(stderr, "nng_sub_open: %s\n", nng_strerror(rv));
return 1;
}
// create AIO
nng_aio_alloc(&node.aio_recv, recv_cb, NULL);
if (rv != 0) {
fprintf(stderr, "nng alloc failed %s\n", nng_strerror(rv) );
}
nng_sub0_socket_subscribe(node.sock, "", 0);
rv = nng_dial(node.sock, argv[1], NULL, 0);
if (rv != 0) {
fprintf(stderr, "nng dial [%s] failed %s\n", argv[1], nng_strerror(rv) );
}
// recv
nng_socket_recv(node.sock, node.aio_recv);
nng_aio_free(node.aio_recv);
nng_socket_close(node.sock);
return 0;
}
`
When the pubsub pattern is used to transmit asynchronous APIs via UDP, messages may sometimes fail to be received. The specific behavior is as follows: the message cannot be received at the current time, but when another message is sent next time, the message sent last time can be received.
Using the following demo, when I send 3 messages from the pub side at a relatively fast speed, the sub side can sometimes only receive 1 message.
Could you please help analyze whether this issue is caused by my improper usage or a problem with the protocol stack? Thank you very much.
// pub.c
`
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <nng/nng.h>
#define SERVER_URL "udp://192.168.31.3:43017"
#define MESSAGE_COUNT 10
int main() {
nng_socket sock;
int ret;
}
`
// sub.c
`
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <nng/nng.h>
void recv_cb(void *arg);
typedef struct {
nng_socket sock;
nng_aio *aio_recv;
} node_t;
node_t node;
// The sub has only received one message.
void recv_cb(void *arg) {
printf("app recv msg !!!\n");
}
int main(int argc, char **argv) {
if (argc < 2) {
fprintf(stderr, "%s \n", argv[0]);
return 0;
}
}
`