summaryrefslogtreecommitdiff
path: root/src/pgm_receiver.hpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/pgm_receiver.hpp')
-rw-r--r--src/pgm_receiver.hpp32
1 files changed, 19 insertions, 13 deletions
diff --git a/src/pgm_receiver.hpp b/src/pgm_receiver.hpp
index 1b367bf..627a658 100644
--- a/src/pgm_receiver.hpp
+++ b/src/pgm_receiver.hpp
@@ -34,7 +34,7 @@
#include "io_object.hpp"
#include "i_engine.hpp"
#include "options.hpp"
-#include "zmq_decoder.hpp"
+#include "decoder.hpp"
#include "pgm_socket.hpp"
namespace zmq
@@ -51,36 +51,42 @@ namespace zmq
int init (bool udp_encapsulation_, const char *network_);
// i_engine interface implementation.
- void plug (struct i_inout *inout_);
+ void plug (class io_thread_t *io_thread_, struct i_inout *inout_);
void unplug ();
- void revive ();
- void resume_input ();
+ void terminate ();
+ void activate_in ();
+ void activate_out ();
// i_poll_events interface implementation.
void in_event ();
+ void timer_event (int token);
private:
+ // RX timeout timer ID.
+ enum {rx_timer_id = 0xa1};
+
+ // RX timer is running.
+ bool has_rx_timer;
+
// If joined is true we are already getting messages from the peer.
// It it's false, we are getting data but still we haven't seen
// beginning of a message.
struct peer_info_t
{
bool joined;
- zmq_decoder_t *decoder;
+ decoder_t *decoder;
};
struct tsi_comp
{
- inline bool operator () (const pgm_tsi_t &ltsi,
+ bool operator () (const pgm_tsi_t &ltsi,
const pgm_tsi_t &rtsi) const
{
- if (ltsi.sport < rtsi.sport)
- return true;
-
- return (std::lexicographical_compare (ltsi.gsi.identifier,
- ltsi.gsi.identifier + 6,
- rtsi.gsi.identifier, rtsi.gsi.identifier + 6));
+ uint32_t ll[2], rl[2];
+ memcpy (ll, &ltsi, sizeof (ll));
+ memcpy (rl, &rtsi, sizeof (rl));
+ return (ll[0] < rl[0]) || (ll[0] == rl[0] && ll[1] < rl[1]);
}
};
@@ -97,7 +103,7 @@ namespace zmq
i_inout *inout;
// Most recently used decoder.
- zmq_decoder_t *mru_decoder;
+ decoder_t *mru_decoder;
// Number of bytes not consumed by the decoder due to pipe overflow.
size_t pending_bytes;