/usr/include/ace/TMCast/LinkListener.hpp is in libace-tmcast-dev 6.0.1-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | // $Id: LinkListener.hpp 91626 2010-09-07 10:59:20Z johnnyw $
// author : Boris Kolpackov <boris@dre.vanderbilt.edu>
// OS primitives
#include <ace/OS_NS_string.h>
#include <ace/OS_NS_stdlib.h>
#include <ace/Synch.h>
#include <ace/SOCK_Dgram_Mcast.h>
#include <ace/Refcounted_Auto_Ptr.h>
#include "Messaging.hpp"
#include "Protocol.hpp"
namespace ACE_TMCast
{
class LinkFailure : public virtual Message {};
class LinkData : public virtual Message
{
public:
LinkData (Protocol::MessageHeader const* header,
void* payload,
size_t size)
: size_ (size)
{
ACE_OS::memcpy (&header_, header, sizeof (Protocol::MessageHeader));
ACE_OS::memcpy (payload_, payload, size_);
}
Protocol::MessageHeader const&
header () const
{
return header_;
}
void const*
payload () const
{
return payload_;
}
size_t
size () const
{
return size_;
}
private:
Protocol::MessageHeader header_;
char payload_[Protocol::MAX_MESSAGE_SIZE];
size_t size_;
};
typedef
ACE_Refcounted_Auto_Ptr<LinkData, ACE_Null_Mutex>
LinkDataPtr;
class LinkListener
{
private:
class Terminate : public virtual Message {};
public:
LinkListener (ACE_SOCK_Dgram_Mcast& sock, MessageQueue& out)
: sock_(sock), out_ (out)
{
ACE_thread_t unused;
if (ACE_OS::thr_create (&thread_thunk,
this,
THR_JOINABLE,
&unused,
&thread_) != 0) ACE_OS::abort ();
}
~LinkListener ()
{
{
MessageQueueAutoLock lock (control_);
control_.push (MessagePtr (new Terminate));
}
if (ACE_OS::thr_join (thread_, 0) != 0) ACE_OS::abort ();
// cerr << "Link listener is down." << endl;
}
static ACE_THR_FUNC_RETURN
thread_thunk (void* arg)
{
LinkListener* obj = reinterpret_cast<LinkListener*> (arg);
obj->execute ();
return 0;
}
void
execute ()
{
char msg[Protocol::MAX_MESSAGE_SIZE];
ssize_t header_size = sizeof (Protocol::MessageHeader);
// OS::Time timeout (1000000); // one millisecond
ACE_Time_Value timeout (0, 1000); // one millisecond
try
{
while (true)
{
// Check control message queue
{
MessageQueueAutoLock lock (control_);
if (!control_.empty ()) break;
}
ACE_INET_Addr junk;
ssize_t n = sock_.recv (msg,
Protocol::MAX_MESSAGE_SIZE,
junk,
0,
&timeout);
if (n != -1)
{
if (n < header_size) throw false;
Protocol::MessageHeader* header =
reinterpret_cast<Protocol::MessageHeader*> (msg);
MessageQueueAutoLock lock (out_);
out_.push (MessagePtr (new LinkData (header,
msg + header_size,
n - header_size)));
}
}
}
catch (...)
{
MessageQueueAutoLock lock (out_);
out_.push (MessagePtr (new LinkFailure));
}
}
private:
// FUZZ: disable check_for_ACE_Guard
typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
// FUZZ: enable check_for_ACE_Guard
ACE_hthread_t thread_;
ACE_SOCK_Dgram_Mcast& sock_;
MessageQueue& out_;
MessageQueue control_;
};
}
|