From cb1b6fe32cbf3c7cf5961bb4156f2de743693a3a Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Mon, 21 Sep 2009 14:39:59 +0200 Subject: initial version of req/rep sockets --- src/rep.cpp | 204 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 204 insertions(+) create mode 100644 src/rep.cpp (limited to 'src/rep.cpp') diff --git a/src/rep.cpp b/src/rep.cpp new file mode 100644 index 0000000..2fbb66c --- /dev/null +++ b/src/rep.cpp @@ -0,0 +1,204 @@ +/* + Copyright (c) 2007-2009 FastMQ Inc. + + This file is part of 0MQ. + + 0MQ is free software; you can redistribute it and/or modify it under + the terms of the Lesser GNU 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 + Lesser GNU General Public License for more details. + + You should have received a copy of the Lesser GNU General Public License + along with this program. If not, see . +*/ + +#include "../bindings/c/zmq.h" + +#include "rep.hpp" +#include "err.hpp" +#include "pipe.hpp" + +zmq::rep_t::rep_t (class app_thread_t *parent_) : + socket_base_t (parent_, ZMQ_REP), + active (0), + current (0), + waiting_for_reply (false), + reply_pipe (NULL) +{ +} + +zmq::rep_t::~rep_t () +{ +} + +bool zmq::rep_t::xrequires_in () +{ + return true; +} + +bool zmq::rep_t::xrequires_out () +{ + return true; +} + +void zmq::rep_t::xattach_pipes (class reader_t *inpipe_, + class writer_t *outpipe_) +{ + zmq_assert (inpipe_ && outpipe_); + zmq_assert (in_pipes.size () == out_pipes.size ()); + + in_pipes.push_back (inpipe_); + in_pipes.swap (active, in_pipes.size () - 1); + out_pipes.push_back (outpipe_); + out_pipes.swap (active, out_pipes.size () - 1); + active++; +} + +void zmq::rep_t::xdetach_inpipe (class reader_t *pipe_) +{ + zmq_assert (pipe_); + zmq_assert (in_pipes.size () == out_pipes.size ()); + + in_pipes_t::size_type index = in_pipes.index (pipe_); + + // If corresponding outpipe is still in place simply nullify the pointer + // to the inpipe and move it to the passive state. + if (out_pipes [index]) { + in_pipes [index] = NULL; + if (in_pipes.index (pipe_) < active) { + active--; + in_pipes.swap (index, active); + out_pipes.swap (index, active); + } + return; + } + + // Now both inpipe and outpipe are detached. Remove them from the lists. + if (in_pipes.index (pipe_) < active) + active--; + in_pipes.erase (index); + out_pipes.erase (index); +} + +void zmq::rep_t::xdetach_outpipe (class writer_t *pipe_) +{ + zmq_assert (pipe_); + zmq_assert (in_pipes.size () == out_pipes.size ()); + + out_pipes_t::size_type index = out_pipes.index (pipe_); + + // TODO: If the connection we've got the request from disconnects, + // there's nowhere to send the reply. DLQ? + if (waiting_for_reply && pipe_ == reply_pipe) { + zmq_assert (false); + } + + // If corresponding inpipe is still in place simply nullify the pointer + // to the outpipe. + if (in_pipes [index]) { + out_pipes [index] = NULL; + if (out_pipes.index (pipe_) < active) { + active--; + in_pipes.swap (index, active); + out_pipes.swap (index, active); + } + return; + } + + // Now both inpipe and outpipe are detached. Remove them from the lists. + if (out_pipes.index (pipe_) < active) + active--; + in_pipes.erase (index); + out_pipes.erase (index); +} + +void zmq::rep_t::xkill (class reader_t *pipe_) +{ + // Move the pipe to the list of inactive pipes. + in_pipes_t::size_type index = in_pipes.index (pipe_); + active--; + in_pipes.swap (index, active); + out_pipes.swap (index, active); +} + +void zmq::rep_t::xrevive (class reader_t *pipe_) +{ + // Move the pipe to the list of active pipes. + in_pipes_t::size_type index = in_pipes.index (pipe_); + in_pipes.swap (index, active); + out_pipes.swap (index, active); + active++; +} + +int zmq::rep_t::xsetsockopt (int option_, const void *optval_, + size_t optvallen_) +{ + errno = EINVAL; + return -1; +} + +int zmq::rep_t::xsend (struct zmq_msg_t *msg_, int flags_) +{ + if (!waiting_for_reply) { + errno = EFAULT; + return -1; + } + + // TODO: Implement this once queue limits are in-place. If the reply + // overloads the buffer, connection should be torn down. + zmq_assert (reply_pipe->check_write (zmq_msg_size (msg_))); + + // Push message to the selected pipe. + reply_pipe->write (msg_); + reply_pipe->flush (); + + waiting_for_reply = false; + reply_pipe = NULL; + + // Detach the message from the data buffer. + int rc = zmq_msg_init (msg_); + zmq_assert (rc == 0); +} + +int zmq::rep_t::xflush () +{ + errno = EFAULT; + return -1; +} + +int zmq::rep_t::xrecv (struct zmq_msg_t *msg_, int flags_) +{ + // Deallocate old content of the message. + zmq_msg_close (msg_); + + if (waiting_for_reply) { + errno = EFAULT; + return -1; + } + + // Round-robin over the pipes to get next message. + for (int count = active; count != 0; count--) { + bool fetched = in_pipes [current]->read (msg_); + current++; + if (current >= active) + current = 0; + if (fetched) { + reply_pipe = out_pipes [current]; + waiting_for_reply = true; + return 0; + } + } + + // No message is available. Initialise the output parameter + // to be a 0-byte message. + zmq_msg_init (msg_); + errno = EAGAIN; + return -1; +} + + -- cgit v1.2.3