Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

* FIX [mqtt] move PINGRESP to protocol layer #959

Merged
merged 4 commits into from
May 22, 2024
Merged

* FIX [mqtt] move PINGRESP to protocol layer #959

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
88144bc
* FIX [mqtt] move PINGRESP to protocol layer
JaylinYu May 21, 2024
1995b4a
* FIX [nmq_mqtt] free ping msg in sock_fini
JaylinYu May 22, 2024
17fc743
* MDF [broker_tcp] fix double close of tcp pipe in rpaio send_cb
JaylinYu May 22, 2024
fc99a0f
* MDF [broker_tls] do the same for mqtt tls trasnport/move pingreq logic
JaylinYu May 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion src/sp/protocol/mqtt/nmq_mqtt.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ struct nano_ctx {
// nano_sock is our per-socket protocol private structure.
struct nano_sock {
nni_mtx lk;
nni_msg *pingmsg;
nni_atomic_int ttl;
nni_id_map pipes;
nni_id_map cached_sessions;
Expand Down Expand Up @@ -523,6 +524,17 @@ nano_sock_init(void *arg, nni_sock *sock)
// Readability comes when there is something on the socket.
nni_pollable_init(&s->writable);
nni_pollable_init(&s->readable);

s->pingmsg = NULL;
nni_msg_alloc(&s->pingmsg, 0);
if (!s->pingmsg) {
log_error("Error in create a pingmsg");
} else {
uint8_t buf[2];
buf[0] = 0xD0;
buf[1] = 0x00;
nni_msg_header_append(s->pingmsg, buf, 2);
}
}

static void
Expand Down Expand Up @@ -1168,8 +1180,21 @@ nano_pipe_recv_cb(void *arg)
case CMD_CONNECT:
case CMD_PUBREC:
case CMD_PUBREL:
case CMD_PINGREQ:
goto drop;
case CMD_PINGREQ:
nni_mtx_lock(&p->lk);
nni_msg_clone(s->pingmsg);
if (!p->busy) {
p->busy = true;
nni_aio_set_msg(&p->aio_send, s->pingmsg);
nni_pipe_send(p->pipe, &p->aio_send);
} else {
if (nni_lmq_put(&p->rlmq, s->pingmsg) != 0) {
nni_msg_free(s->pingmsg);
}
}
nni_mtx_unlock(&p->lk);
break;
default:
goto drop;
}
Expand Down
28 changes: 14 additions & 14 deletions src/sp/transport/mqtt/broker_tcp.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -708,20 +708,20 @@ tcptran_pipe_recv_cb(void *arg)
return;
}
// move to protocol layer
if (p->gotrxhead == 2) {
if ((p->rxlen[0] & 0XFF) == CMD_PINGREQ) {
// TODO set timeout in case it never finish
nng_aio_wait(p->rpaio);
p->txlen[0] = CMD_PINGRESP;
p->txlen[1] = 0x00;
iov[0].iov_len = 2;
iov[0].iov_buf = &p->txlen;
// send CMD_PINGRESP down...
nni_aio_set_iov(p->rpaio, 1, iov);
nng_stream_send(p->conn, p->rpaio);
goto notify;
}
}
// if (p->gotrxhead == 2) {
// if ((p->rxlen[0] & 0XFF) == CMD_PINGREQ) {
// // TODO set timeout in case it never finish
// nng_aio_wait(p->rpaio);
// p->txlen[0] = CMD_PINGRESP;
// p->txlen[1] = 0x00;
// iov[0].iov_len = 2;
// iov[0].iov_buf = &p->txlen;
// // send CMD_PINGRESP down...
// nni_aio_set_iov(p->rpaio, 1, iov);
// nng_stream_send(p->conn, p->rpaio);
// goto notify;
// }
// }

if (p->rxmsg == NULL) {
if ((rv = mqtt_get_remaining_length(
Expand Down
Loading