3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-03-09 15:26:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

ZMQII-5: Only one consumer in a process gets the message

Browse Source
This commit is contained in:
Martin Sustrik 2009-09-11 13:20:24 +02:00
parent f824b8a067
commit 5daa0dec0f
4 changed files with 22 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
src/session.cpp
View File

@ -72,9 +72,9 @@ void zmq::session_t::detach ()
// Engine is terminating itself.
engine = NULL;
// TODO: In the case od anonymous connection, terminate the session.
// if (anonymous)
// term ();
// In the case od anonymous connection, terminate the session.
if (name.empty ())
term ();
}
void zmq::session_t::attach_inpipe (reader_t *pipe_)
@ -114,11 +114,13 @@ void zmq::session_t::detach_outpipe (writer_t *pipe_)
void zmq::session_t::process_plug ()
{
// Register the session with the socket.
bool ok = owner->register_session (name.c_str (), this);
if (!name.empty ()) {
bool ok = owner->register_session (name.c_str (), this);
// There's already a session with the specified identity.
// We should syslog it and drop the session. TODO
zmq_assert (ok);
// There's already a session with the specified identity.
// We should syslog it and drop the session. TODO
zmq_assert (ok);
}
// If session is created by 'connect' function, it has the pipes set
// already. Otherwise, it's being created by the listener and the pipes
@ -141,8 +143,10 @@ void zmq::session_t::process_plug ()
void zmq::session_t::process_unplug ()
{
// Unregister the session from the socket.
bool ok = owner->unregister_session (name.c_str ());
zmq_assert (ok);
if (!name.empty ()) {
bool ok = owner->unregister_session (name.c_str ());
zmq_assert (ok);
}
// Ask associated pipes to terminate.
if (in_pipe) {
@ -163,6 +167,7 @@ void zmq::session_t::process_unplug ()
void zmq::session_t::process_attach (i_engine *engine_)
{
zmq_assert (!engine);
zmq_assert (engine_);
engine = engine_;
engine->plug (this);

4
src/zmq_connecter_init.cpp
View File

@ -55,7 +55,9 @@ bool zmq::zmq_connecter_init_t::read (::zmq_msg_t *msg_)
// Find the session associated with this connecter. If it doesn't exist
// drop the newly created connection. If it does, attach it to the
// connection.
session_t *session = owner->find_session (session_name.c_str ());
session_t *session = NULL;
if (!session_name.empty ())
session = owner->find_session (session_name.c_str ());
if (!session) {
// TODO
zmq_assert (false);

2
src/zmq_engine.cpp
View File

@ -50,6 +50,8 @@ zmq::zmq_engine_t::~zmq_engine_t ()
void zmq::zmq_engine_t::plug (i_inout *inout_)
{
zmq_assert (!inout);
encoder.set_inout (inout_);
decoder.set_inout (inout_);

4
src/zmq_listener_init.cpp
View File

@ -70,7 +70,9 @@ void zmq::zmq_listener_init_t::flush ()
// Have a look whether the session already exists. If it does, attach it
// to the engine. If it doesn't create it first.
session_t *session = owner->find_session (peer_identity.c_str ());
session_t *session = NULL;
if (!peer_identity.empty ())
session = owner->find_session (peer_identity.c_str ());
if (!session) {
io_thread_t *io_thread = choose_io_thread (options.affinity);
session = new session_t (io_thread, owner, peer_identity.c_str (),