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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
|
/* -*- C++ -*- */
// $Id$
// ============================================================================
//
// = LIBRARY
// apps
//
// = FILENAME
// IO_Handler.h
//
// = AUTHOR
// Doug Schmidt
//
// ============================================================================
#if !defined (_IO_HANDLER)
#define _IO_HANDLER
#include "ace/Service_Config.h"
#include "ace/SOCK_Connector.h"
#include "ace/Svc_Handler.h"
#include "Consumer_Map.h"
#include "Consumer_Entry.h"
#include "Event.h"
// Forward declaration.
class IO_Handler_Connector;
class IO_Handler : public ACE_Svc_Handler<ACE_SOCK_STREAM, SYNCH_STRATEGY>
// = TITLE
// IO_Handler contains info about connection state and addressing.
//
// = DESCRIPTION
// The IO_Handler classes process events sent from the peers to the
// gateway. These classes works as follows:
//
// 1. IO_Handler_Connector creates a number of connections with the set of
// peers specified in a configuration file.
//
// 2. For each peer that connects successfully, IO_Handler_Connector
// creates an IO_Handler object. Each object assigns a unique routing
// id to its associated peer. The Handlers are used by gatewayd
// that to receive, route, and forward events from source peer(s)
// to destination peer(s).
{
public:
IO_Handler (Consumer_Map *,
IO_Handler_Connector *,
ACE_Thread_Manager * = 0,
int socket_queue_size = 0);
virtual int open (void * = 0);
// Initialize and activate a single-threaded IO_Handler (called by
// ACE_Connector::handle_output()).
int bind (const ACE_INET_Addr &remote_addr,
const ACE_INET_Addr &local_addr,
CONN_ID);
// Set the peer's addressing and routing information.
ACE_INET_Addr &remote_addr (void);
// Returns the peer's routing address.
ACE_INET_Addr &local_addr (void);
// Returns our local address.
// = Set/get routing id.
CONN_ID id (void);
void id (CONN_ID);
// = Set/get the current state of the IO_Handler.
enum State
{
IDLE = 1, // Prior to initialization.
CONNECTING, // During connection establishment.
ESTABLISHED, // IO_Handler is established and active.
DISCONNECTING, // IO_Handler is in the process of connecting.
FAILED // IO_Handler has failed.
};
// = Set/get the current state.
State state (void);
void state (State);
// = Set/get the current retry timeout delay.
int timeout (void);
void timeout (int);
// = Set/get the maximum retry timeout delay.
int max_timeout (void);
void max_timeout (int);
// = Set/get IO_Handler activity status.
int active (void);
void active (int);
// = Set/get direction (necessary for error checking).
char direction (void);
void direction (char);
// = The total number of bytes sent/received on this channel.
size_t total_bytes (void);
void total_bytes (size_t bytes);
// Increment count by <bytes>.
virtual int handle_timeout (const ACE_Time_Value &, const void *arg);
// Perform timer-based IO_Handler reconnection.
protected:
enum
{
MAX_RETRY_TIMEOUT = 300 // 5 minutes is the maximum timeout.
};
int initialize_connection (void);
// Perform the first-time initiation of a connection to the peer.
int reinitiate_connection (void);
// Reinitiate a connection asynchronously when peers fail.
void socket_queue_size (void);
// Set the socket queue size.
virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,
ACE_Reactor_Mask = ACE_Event_Handler::RWE_MASK);
// Perform IO_Handler termination.
Consumer_Map *consumer_map_;
// Pointer to table that maps an event
// to a Set of IO_Handler *'s for output.
ACE_INET_Addr remote_addr_;
// Address of peer.
ACE_INET_Addr local_addr_;
// Address of us.
CONN_ID id_;
// The assigned routing ID of this entry.
size_t total_bytes_;
// The total number of bytes sent/received on this channel.
State state_;
// The current state of the channel.
IO_Handler_Connector *connector_;
// Back pointer to IO_Handler_Connector to reestablish broken
// connections.
int timeout_;
// Amount of time to wait between reconnection attempts.
int max_timeout_;
// Maximum amount of time to wait between reconnection attempts.
char direction_;
// Indicates which direction data flows through the channel ('O' ==
// output and 'I' == input).
int socket_queue_size_;
// Size of the socket queue (0 means "use default").
};
class Supplier_Handler : public IO_Handler
// = TITLE
// Handle reception of Peer events arriving as events.
{
public:
Supplier_Handler (Consumer_Map *,
IO_Handler_Connector *,
ACE_Thread_Manager * = 0,
int socket_queue_size = 0);
// Constructor sets the consumer map pointer.
virtual int handle_input (ACE_HANDLE = ACE_INVALID_HANDLE);
// Receive and process peer events.
protected:
virtual int recv (ACE_Message_Block *&);
// Receive an event from a Supplier.
int forward (ACE_Message_Block *event);
// Forward the Event to a Consumer.
ACE_Message_Block *msg_frag_;
// Keep track of event fragment to handle non-blocking recv's from
// Suppliers.
};
class Consumer_Handler : public IO_Handler
// = TITLE
// Handle transmission of events to other Peers using a
// single-threaded approach.
{
public:
Consumer_Handler (Consumer_Map *,
IO_Handler_Connector *,
ACE_Thread_Manager * = 0,
int socket_queue_size = 0);
virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0);
// Send an event to a Consumer (may be queued if necessary).
protected:
// = We'll allow up to 16 megabytes to be queued per-output
// channel.
enum {QUEUE_SIZE = 1024 * 1024 * 16};
virtual int handle_input (ACE_HANDLE);
// Receive and process shutdowns from a Consumer.
virtual int handle_output (ACE_HANDLE);
// Finish sending event when flow control conditions abate.
int nonblk_put (ACE_Message_Block *mb);
// Perform a non-blocking put().
virtual ssize_t send (ACE_Message_Block *);
// Send an event to a Consumer.
};
#endif /* _IO_HANDLER */
|
