summaryrefslogtreecommitdiff
path: root/src/pgm_sender.cpp
diff options
context:
space:
mode:
authorMartin Lucina <martin@lucina.net>2012-01-23 08:53:35 +0100
committerMartin Lucina <martin@lucina.net>2012-01-23 08:53:35 +0100
commite645fc2693acc796304498909786b7b47005b429 (patch)
tree4118cd4c7b9eba3ba1d6892800c79669ea94c4e9 /src/pgm_sender.cpp
parent2c416a793ea781273a5da6742211f5f01af13a2b (diff)
Imported Upstream version 2.1.3upstream/2.1.3
Diffstat (limited to 'src/pgm_sender.cpp')
-rw-r--r--src/pgm_sender.cpp77
1 files changed, 66 insertions, 11 deletions
diff --git a/src/pgm_sender.cpp b/src/pgm_sender.cpp
index 9aeb7a9..4d76433 100644
--- a/src/pgm_sender.cpp
+++ b/src/pgm_sender.cpp
@@ -1,19 +1,20 @@
/*
- Copyright (c) 2007-2010 iMatix Corporation
+ 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 Lesser GNU General Public License as published by
+ 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
- Lesser GNU General Public License for more details.
+ GNU Lesser General Public License for more details.
- You should have received a copy of the Lesser GNU General Public License
+ 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/>.
*/
@@ -36,6 +37,8 @@
zmq::pgm_sender_t::pgm_sender_t (io_thread_t *parent_,
const options_t &options_) :
io_object_t (parent_),
+ has_tx_timer (false),
+ has_rx_timer (false),
encoder (0),
pgm_socket (false, options_),
options (options_),
@@ -53,12 +56,12 @@ int zmq::pgm_sender_t::init (bool udp_encapsulation_, const char *network_)
out_buffer_size = pgm_socket.get_max_tsdu_size ();
out_buffer = (unsigned char*) malloc (out_buffer_size);
- zmq_assert (out_buffer);
+ alloc_assert (out_buffer);
return rc;
}
-void zmq::pgm_sender_t::plug (i_inout *inout_)
+void zmq::pgm_sender_t::plug (io_thread_t *io_thread_, i_inout *inout_)
{
// Alocate 2 fds for PGM socket.
int downlink_socket_fd = 0;
@@ -89,6 +92,16 @@ void zmq::pgm_sender_t::plug (i_inout *inout_)
void zmq::pgm_sender_t::unplug ()
{
+ if (has_rx_timer) {
+ cancel_timer (rx_timer_id);
+ has_rx_timer = false;
+ }
+
+ if (has_tx_timer) {
+ cancel_timer (tx_timer_id);
+ has_tx_timer = false;
+ }
+
rm_fd (handle);
rm_fd (uplink_handle);
rm_fd (rdata_notify_handle);
@@ -96,13 +109,19 @@ void zmq::pgm_sender_t::unplug ()
encoder.set_inout (NULL);
}
-void zmq::pgm_sender_t::revive ()
+void zmq::pgm_sender_t::terminate ()
+{
+ unplug ();
+ delete this;
+}
+
+void zmq::pgm_sender_t::activate_out ()
{
set_pollout (handle);
out_event ();
}
-void zmq::pgm_sender_t::resume_input ()
+void zmq::pgm_sender_t::activate_in ()
{
zmq_assert (false);
}
@@ -117,8 +136,18 @@ zmq::pgm_sender_t::~pgm_sender_t ()
void zmq::pgm_sender_t::in_event ()
{
- // In event on sender side means NAK or SPMR receiving from some peer.
+ if (has_rx_timer) {
+ cancel_timer (rx_timer_id);
+ has_rx_timer = false;
+ }
+
+ // In-event on sender side means NAK or SPMR receiving from some peer.
pgm_socket.process_upstream ();
+ if (errno == ENOMEM || errno == EBUSY) {
+ const long timeout = pgm_socket.get_rx_timeout ();
+ add_timer (timeout, rx_timer_id);
+ has_rx_timer = true;
+ }
}
void zmq::pgm_sender_t::out_event ()
@@ -146,14 +175,40 @@ void zmq::pgm_sender_t::out_event ()
put_uint16 (out_buffer, offset == -1 ? 0xffff : (uint16_t) offset);
}
+ if (has_tx_timer) {
+ cancel_timer (tx_timer_id);
+ has_tx_timer = false;
+ }
+
// Send the data.
size_t nbytes = pgm_socket.send (out_buffer, write_size);
// We can write either all data or 0 which means rate limit reached.
- if (nbytes == write_size)
+ if (nbytes == write_size) {
write_size = 0;
- else
+ } else {
zmq_assert (nbytes == 0);
+
+ if (errno == ENOMEM) {
+ const long timeout = pgm_socket.get_tx_timeout ();
+ add_timer (timeout, tx_timer_id);
+ has_tx_timer = true;
+ } else
+ zmq_assert (errno == EBUSY);
+ }
+}
+
+void zmq::pgm_sender_t::timer_event (int token)
+{
+ // Timer cancels on return by poller_base.
+ if (token == rx_timer_id) {
+ has_rx_timer = false;
+ in_event ();
+ } else if (token == tx_timer_id) {
+ has_tx_timer = false;
+ out_event ();
+ } else
+ zmq_assert (false);
}
#endif