Skip to content

L
libzmq
Commit dee8b236 authored by Martin Sustrik's avatar Martin Sustrik
Browse files
Options

New style ROUTER socket removed. 

Signed-off-by: 's avatarMartin Sustrik <sustrik@turist.(none)>
parent 541b83bc
Show whitespace changes
Inline Side-by-side
Showing with 0 additions and 420 deletions
include/zmq.h
View file @ dee8b236
......@@ -162,7 +162,6 @@ ZMQ_EXPORT int zmq_term (void *context);
#define ZMQ_PUSH 8
#define ZMQ_XPUB 9
#define ZMQ_XSUB 10
#define ZMQ_ROUTER 13
/* Socket options. */
#define ZMQ_AFFINITY 4
......
src/Makefile.am
View file @ dee8b236
......@@ -55,7 +55,6 @@ libzmq_la_SOURCES = \
reaper.hpp \
rep.hpp \
req.hpp \
router.hpp \
select.hpp \
session_base.hpp \
signaler.hpp \
......@@ -113,7 +112,6 @@ libzmq_la_SOURCES = \
reaper.cpp \
pub.cpp \
random.cpp \
router.cpp \
rep.cpp \
req.cpp \
select.cpp \
......
src/router.cpp deleted 100755 → 0
View file @ 541b83bc
/*
Copyright (c) 2007-2011 iMatix Corporation
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "router.hpp"
#include "pipe.hpp"
#include "wire.hpp"
#include "random.hpp"
#include "likely.hpp"
#include "wire.hpp"
#include "err.hpp"
zmq::router_t::router_t (class ctx_t *parent_, uint32_t tid_) :
socket_base_t (parent_, tid_),
prefetched (false),
more_in (false),
current_out (NULL),
more_out (false),
next_peer_id (generate_random ())
{
options.type = ZMQ_ROUTER;
prefetched_msg.init ();
}
zmq::router_t::~router_t ()
{
zmq_assert (outpipes.empty ());
prefetched_msg.close ();
}
void zmq::router_t::xattach_pipe (pipe_t *pipe_)
{
zmq_assert (pipe_);
// Generate a new peer ID. Take care to avoid duplicates.
outpipes_t::iterator it = outpipes.lower_bound (next_peer_id);
if (!outpipes.empty ()) {
while (true) {
if (it == outpipes.end ())
it = outpipes.begin ();
if (it->first != next_peer_id)
break;
++next_peer_id;
++it;
}
}
// Add the pipe to the map out outbound pipes.
outpipe_t outpipe = {pipe_, true};
bool ok = outpipes.insert (outpipes_t::value_type (
next_peer_id, outpipe)).second;
zmq_assert (ok);
// Add the pipe to the list of inbound pipes.
pipe_->set_pipe_id (next_peer_id);
fq.attach (pipe_);
// Queue the connection command.
pending_command_t cmd = {1, next_peer_id};
pending_commands.push_back (cmd);
// Advance next peer ID so that if new connection is dropped shortly after
// its creation we don't accidentally get two subsequent peers with
// the same ID.
++next_peer_id;
}
void zmq::router_t::xterminated (pipe_t *pipe_)
{
fq.terminated (pipe_);
for (outpipes_t::iterator it = outpipes.begin ();
it != outpipes.end (); ++it) {
if (it->second.pipe == pipe_) {
// Queue the disconnection command.
pending_command_t cmd = {2, it->first};
pending_commands.push_back (cmd);
// Remove the pipe.
outpipes.erase (it);
if (pipe_ == current_out)
current_out = NULL;
return;
}
}
zmq_assert (false);
}
void zmq::router_t::xread_activated (pipe_t *pipe_)
{
fq.activated (pipe_);
}
void zmq::router_t::xwrite_activated (pipe_t *pipe_)
{
for (outpipes_t::iterator it = outpipes.begin ();
it != outpipes.end (); ++it) {
if (it->second.pipe == pipe_) {
zmq_assert (!it->second.active);
it->second.active = true;
return;
}
}
zmq_assert (false);
}
int zmq::router_t::xsend (msg_t *msg_, int flags_)
{
// If this is the first part of the message it's the ID of the
// peer to send the message to.
if (!more_out) {
zmq_assert (!current_out);
// The first message part has to be label.
if (unlikely (!(msg_->flags () & msg_t::label))) {
errno = EFSM;
return -1;
}
// Find the pipe associated with the peer ID stored in the message.
if (unlikely (msg_->size () != 4)) {
errno = ECANTROUTE;
return -1;
}
uint32_t peer_id = get_uint32 ((unsigned char*) msg_->data ());
outpipes_t::iterator it = outpipes.find (peer_id);
if (unlikely (it == outpipes.end ())) {
errno = ECANTROUTE;
return -1;
}
// Check whether the pipe is available for writing.
msg_t empty;
int rc = empty.init ();
errno_assert (rc == 0);
if (!it->second.pipe->check_write (&empty)) {
rc = empty.close ();
errno_assert (rc == 0);
it->second.active = false;
errno = EAGAIN;
return -1;
}
rc = empty.close ();
errno_assert (rc == 0);
// Mark the pipe to send the message to.
current_out = it->second.pipe;
more_out = true;
// Clean up the message object.
rc = msg_->close ();
errno_assert (rc == 0);
rc = msg_->init ();
errno_assert (rc == 0);
return 0;
}
// Check whether this is the last part of the message.
more_out = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
// Push the message into the pipe. If there's no out pipe, just drop it.
if (current_out) {
bool ok = current_out->write (msg_);
if (unlikely (!ok))
current_out = NULL;
else if (!more_out) {
current_out->flush ();
current_out = NULL;
}
}
else {
int rc = msg_->close ();
errno_assert (rc == 0);
}
// Detach the message from the data buffer.
int rc = msg_->init ();
errno_assert (rc == 0);
return 0;
}
int zmq::router_t::xrecv (msg_t *msg_, int flags_)
{
// If there's a queued command, pass it to the caller.
if (unlikely (!more_in && !pending_commands.empty ())) {
msg_->init_size (5);
unsigned char *data = (unsigned char*) msg_->data ();
put_uint8 (data, pending_commands.front ().cmd);
put_uint32 (data + 1, pending_commands.front ().peer);
msg_->set_flags (msg_t::command);
pending_commands.pop_front ();
return 0;
}
// If there is a prefetched message, return it.
if (prefetched) {
int rc = msg_->move (prefetched_msg);
errno_assert (rc == 0);
more_in = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
prefetched = false;
return 0;
}
// Get next message part.
pipe_t *pipe;
int rc = fq.recvpipe (msg_, flags_, &pipe);
if (rc != 0)
return -1;
// If we are in the middle of reading a message, just return the next part.
if (more_in) {
more_in = msg_->flags () & (msg_t::more | msg_t::label) ? true : false;
return 0;
}
// We are at the beginning of a new message. Move the message part we
// have to the prefetched and return the ID of the peer instead.
rc = prefetched_msg.move (*msg_);
errno_assert (rc == 0);
prefetched = true;
rc = msg_->close ();
errno_assert (rc == 0);
rc = msg_->init_size (4);
errno_assert (rc == 0);
put_uint32 ((unsigned char*) msg_->data (), pipe->get_pipe_id ());
msg_->set_flags (msg_t::label);
return 0;
}
int zmq::router_t::rollback (void)
{
if (current_out) {
current_out->rollback ();
current_out = NULL;
more_out = false;
}
return 0;
}
bool zmq::router_t::xhas_in ()
{
if (prefetched)
return true;
return fq.has_in () || !pending_commands.empty();
}
bool zmq::router_t::xhas_out ()
{
// In theory, GENERIC socket is always ready for writing. Whether actual
// attempt to write succeeds depends on whitch pipe the message is going
// to be routed to.
return true;
}
zmq::router_session_t::router_session_t (io_thread_t *io_thread_, bool connect_,
socket_base_t *socket_, const options_t &options_,
const char *protocol_, const char *address_) :
session_base_t (io_thread_, connect_, socket_, options_, protocol_,
address_)
{
}
zmq::router_session_t::~router_session_t ()
{
}
src/router.hpp deleted 100755 → 0
View file @ 541b83bc
/*
Copyright (c) 2007-2011 iMatix Corporation
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
0MQ is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_ROUTER_HPP_INCLUDED__
#define __ZMQ_ROUTER_HPP_INCLUDED__
#include <map>
#include <deque>
#include "socket_base.hpp"
#include "session_base.hpp"
#include "stdint.hpp"
#include "msg.hpp"
#include "fq.hpp"
namespace zmq
{
class router_t :
public socket_base_t
{
public:
router_t (class ctx_t *parent_, uint32_t tid_);
~router_t ();
// Overloads of functions from socket_base_t.
void xattach_pipe (class pipe_t *pipe_);
int xsend (class msg_t *msg_, int flags_);
int xrecv (class msg_t *msg_, int flags_);
bool xhas_in ();
bool xhas_out ();
void xread_activated (class pipe_t *pipe_);
void xwrite_activated (class pipe_t *pipe_);
void xterminated (class pipe_t *pipe_);
protected:
// Rollback any message parts that were sent but not yet flushed.
int rollback ();
private:
// Fair queueing object for inbound pipes.
fq_t fq;
// Have we prefetched a message.
bool prefetched;
// Holds the prefetched message.
msg_t prefetched_msg;
// If true, more incoming message parts are expected.
bool more_in;
struct outpipe_t
{
class pipe_t *pipe;
bool active;
};
// Outbound pipes indexed by the peer IDs.
typedef std::map <uint32_t, outpipe_t> outpipes_t;
outpipes_t outpipes;
// The pipe we are currently writing to.
class pipe_t *current_out;
// If true, more outgoing message parts are expected.
bool more_out;
// Peer ID are generated. It's a simple increment and wrap-over
// algorithm. This value is the next ID to use (if not used already).
uint32_t next_peer_id;
// Commands to be delivered to the user.
struct pending_command_t
{
uint8_t cmd;
uint32_t peer;
};
typedef std::deque <pending_command_t> pending_commands_t;
pending_commands_t pending_commands;
router_t (const router_t&);
const router_t &operator = (const router_t&);
};
class router_session_t : public session_base_t
{
public:
router_session_t (class io_thread_t *io_thread_, bool connect_,
class socket_base_t *socket_, const options_t &options_,
const char *protocol_, const char *address_);
~router_session_t ();
private:
router_session_t (const router_session_t&);
const router_session_t &operator = (const router_session_t&);
};
}
#endif
src/session_base.cpp
View file @ dee8b236
......@@ -40,7 +40,6 @@
#include "xsub.hpp"
#include "push.hpp"
#include "pull.hpp"
#include "router.hpp"
#include "pair.hpp"
zmq::session_base_t *zmq::session_base_t::create (class io_thread_t *io_thread_,
......@@ -88,10 +87,6 @@ zmq::session_base_t *zmq::session_base_t::create (class io_thread_t *io_thread_,
s = new (std::nothrow) pull_session_t (io_thread_, connect_,
socket_, options_, protocol_, address_);
break;
case ZMQ_ROUTER:
s = new (std::nothrow) router_session_t (io_thread_, connect_,
socket_, options_, protocol_, address_);
break;
case ZMQ_PAIR:
s = new (std::nothrow) pair_session_t (io_thread_, connect_,
socket_, options_, protocol_, address_);
......
src/socket_base.cpp
View file @ dee8b236
......@@ -60,7 +60,6 @@
#include "xrep.hpp"
#include "xpub.hpp"
#include "xsub.hpp"
#include "router.hpp"
bool zmq::socket_base_t::check_tag ()
{
......@@ -106,9 +105,6 @@ zmq::socket_base_t *zmq::socket_base_t::create (int type_, class ctx_t *parent_,
case ZMQ_XSUB:
s = new (std::nothrow) xsub_t (parent_, tid_);
break;
case ZMQ_ROUTER:
s = new (std::nothrow) router_t (parent_, tid_);
break;
default:
errno = EINVAL;
return NULL;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment