generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-13 02:42:58 +01:00
Code Issues Projects Releases Wiki Activity

rollback of half-processed messages in case of disconnection

Browse Source
This commit is contained in:
Martin Sustrik 2010-03-20 15:04:30 +01:00
parent dfdaff5eba
commit f031677100
2 changed files with 30 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/session.cpp
View File

@ -28,6 +28,7 @@ zmq::session_t::session_t (object_t *parent_, socket_base_t *owner_,
const options_t &options_) :
owned_t (parent_, owner_),
in_pipe (NULL),
incomplete_in (false),
active (true),
out_pipe (NULL),
engine (NULL),
@ -72,7 +73,11 @@ bool zmq::session_t::read (::zmq_msg_t *msg_)
if (!in_pipe || !active)
return false;
return in_pipe->read (msg_);
if (!in_pipe->read (msg_))
return false;
incomplete_in = msg_->flags & ZMQ_MSG_TBC;
return true;
}
bool zmq::session_t::write (::zmq_msg_t *msg_)
@ -102,6 +107,26 @@ void zmq::session_t::detach (owned_t *reconnecter_)
// Engine is terminating itself. No need to deallocate it from here.
engine = NULL;
// Get rid of half-processed messages in the out pipe. Flush any
// unflushed messages upstream.
if (out_pipe) {
out_pipe->rollback ();
out_pipe->flush ();
}
// Remove any half-read message from the in pipe.
if (in_pipe) {
while (incomplete_in) {
zmq_msg_t msg;
zmq_msg_init (&msg);
if (!read (&msg)) {
zmq_assert (!incomplete_in);
break;
}
zmq_msg_close (&msg);
}
}
// Terminate transient session.
if (!ordinal && (peer_identity.empty () || peer_identity [0] == 0))
term ();

4
src/session.hpp
View File

@ -72,6 +72,10 @@ namespace zmq
// Inbound pipe, i.e. one the session is getting messages from.
class reader_t *in_pipe;
// This flag is true if the remainder of the message being processed
// is still in the in pipe.
bool incomplete_in;
// If true, in_pipe is active. Otherwise there are no messages to get.
bool active;