summaryrefslogtreecommitdiff
path: root/src/rep.cpp
diff options
context:
space:
mode:
authorMartin Sustrik <sustrik@fastmq.commkdir>2009-09-21 14:39:59 +0200
committerMartin Sustrik <sustrik@fastmq.commkdir>2009-09-21 14:39:59 +0200
commitcb1b6fe32cbf3c7cf5961bb4156f2de743693a3a (patch)
treefad2a866ccb543fd4676c4539c68fb32c52dc3a3 /src/rep.cpp
parent7668b246fc3cf4a2a3b3ee9b1283ad8a4b12ac4f (diff)
initial version of req/rep sockets
Diffstat (limited to 'src/rep.cpp')
-rw-r--r--src/rep.cpp204
1 files changed, 204 insertions, 0 deletions
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 <http://www.gnu.org/licenses/>.
+*/
+
+#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;
+}
+
+