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
|
// $Id$
#include "ace/OS_main.h"
#include "ace/RMCast/RMCast_UDP_Reliable_Sender.h"
#include "ace/INET_Addr.h"
#include "ace/FILE_IO.h"
#include "ace/Message_Block.h"
#include "ace/Reactor.h"
ACE_RCSID(tests, RMCast_Examples_Sender, "$Id$")
int
ACE_TMAIN (int argc, ACE_TCHAR *argv[])
{
if (argc != 3)
{
ACE_ERROR_RETURN ((LM_ERROR,
"Usage: %s <filename> <mcastgroup:port>\n",
argv[0]),
1);
}
const ACE_TCHAR *filename = argv[1];
if (ACE_OS::access (filename, R_OK) != 0)
{
ACE_ERROR_RETURN ((LM_ERROR,
"Cannot read file <%s>\n", filename),
1);
}
ACE_INET_Addr mcast_group;
if (mcast_group.set (argv[2]) != 0)
{
ACE_ERROR_RETURN ((LM_ERROR,
"Cannot setup multicast group <%s>\n",
argv[2]),
1);
}
ACE_HANDLE handle = ACE_OS::open (filename, O_RDONLY|O_BINARY);
if (handle == ACE_INVALID_HANDLE)
{
ACE_ERROR_RETURN ((LM_ERROR,
"Cannot open file <%s> %p\n", filename, ""),
1);
}
ACE_FILE_IO file_io;
file_io.set_handle (handle);
// We don't provide a module to receive the control messages, in
// this example we simply ignore them.
ACE_RMCast_UDP_Reliable_Sender sender (0);
if (sender.init (mcast_group) == -1)
{
ACE_ERROR_RETURN ((LM_ERROR,
"Cannot init UDP I/O at <%s:%d> %p\n",
mcast_group.get_host_name (),
mcast_group.get_port_number (),
""),
1);
}
// Use the Reactor to demultiplex all the messages
ACE_Reactor *reactor = ACE_Reactor::instance ();
sender.reactive_incoming_messages (reactor);
{
// Resend the messages every 20 milliseconds..
ACE_Time_Value tv (2, 0);
sender.reactive_resends (reactor, tv);
}
for (;;)
{
ACE_Message_Block payload (BUFSIZ + 1);
ssize_t r = file_io.recv (payload.rd_ptr () + 1, BUFSIZ);
if (r <= 0)
break;
payload.wr_ptr (r + 1);
*(payload.rd_ptr ()) = 'N'; // Normal
if (r < BUFSIZ)
{
*(payload.rd_ptr ()) = 'E'; // EOF
}
ACE_RMCast::Data data;
data.payload = &payload;
if (sender.data (data) != 0)
break;
if (r < BUFSIZ)
{
// Last buffer, terminate loop
break;
}
// Handle incoming events, without blocking...
ACE_Time_Value tv (4, 0);
reactor->handle_events (&tv);
}
// Wait until all the messages are successfully delivered
do
{
// Try for 50 milliseconds...
ACE_Time_Value tv (5, 0);
int r = reactor->handle_events (&tv);
if (r == -1)
break;
}
while (sender.has_data () || sender.has_members ());
return 0;
}
|
