* MDF [broker] use msg_pipe not aio:pipe
@@ -110,7 +110,6 @@ server_cb(void *arg)
} else if (nng_msg_cmd_type(msg) == CMD_CONNACK) {
work->pid = nng_msg_get_pipe(work->msg);
nng_msg_set_pipe(work->msg, work->pid);
- // nng_aio_set_pipeline(work->aio, work->pid.id);
nng_aio_set_msg(work->aio, work->msg);
work->msg = NULL;
work->state = SEND;
@@ -144,7 +143,6 @@ server_cb(void *arg)
work->msg = smsg;
work->pid = nng_msg_get_pipe(work->msg);
nng_msg_set_pipe(work->msg, work->pid);
- // nng_aio_set_pipeline(work->aio, work->pid.id);
nng_aio_set_msg(work->aio, work->msg);
work->msg = NULL;
work->state = SEND;
@@ -202,7 +200,6 @@ server_cb(void *arg)
work->msg = m;
nng_aio_set_msg(work->aio, work->msg);
nng_msg_set_pipe(work->msg, work->pid);
- // nng_aio_set_pipeline(work->aio, work->pid.id);
nng_ctx_send(work->ctx, work->aio);
}
cvector_free(work->msg_ret);
@@ -210,7 +207,6 @@ server_cb(void *arg)
nng_msg_set_cmd_type(smsg, CMD_SUBACK);
work->msg = smsg;
nng_msg_set_pipe(work->msg, work->pid);
- // nng_aio_set_pipeline(work->aio, work->pid.id);
nng_aio_set_msg(work->aio, work->msg);
work->msg = NULL;
work->state = SEND;
@@ -256,7 +252,8 @@ server_cb(void *arg)
nng_msg_free(work->msg);
work->msg = smsg;
- // We could add more data to the message here.
+ work->pid.id = 0;
+ nng_msg_set_pipe(work->msg, work->pid);
nng_aio_set_msg(work->aio, work->msg);
work->msg = NULL;
work->state = SEND;
@@ -290,13 +287,10 @@ server_cb(void *arg)
nng_msg_clone(smsg);
work->msg = smsg;
nng_aio_set_msg(work->aio, work->msg);
- work->msg = NULL;
-
- if (p_info.pipe != 0) {
+ //TODO pipe = 0?
work->pid.id = p_info.pipe;
nng_msg_set_pipe(work->msg, work->pid);
- // nng_aio_set_pipeline(work->aio, p_info.pipe);
- }
+ work->msg = NULL;
work->state = SEND;
work->pipe_ct->current_index++;
nng_ctx_send(work->ctx, work->aio);
