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
|
/* -*- C++ -*- */
//=============================================================================
/**
* @file Concrete_Connection_Handlers.h
*
* These are all the subclasses of Connection_Handler that define the
* appropriate threaded/reactive Consumer/Supplier behavior.
*
* @author Doug Schmidt <d.schmidt@vanderbilt.edu>
*/
//=============================================================================
#ifndef CONCRETE_CONNECTION_HANDLER
#define CONCRETE_CONNECTION_HANDLER
#include "Connection_Handler.h"
/**
* @class Supplier_Handler
*
* @brief Handles reception of Events from Suppliers.
*
* Performs framing and error checking on Events. Intended to
* run reactively, i.e., in one thread of control using a
* Reactor for demuxing and dispatching.
*/
class Supplier_Handler : public Connection_Handler
{
public:
Supplier_Handler (const Connection_Config_Info &);
protected:
// = All the following methods are upcalls, so they can be protected.
/// Receive and process peer events.
virtual int handle_input (ACE_HANDLE = ACE_INVALID_HANDLE);
/// Receive an event from a Supplier.
virtual int recv (ACE_Message_Block *&);
/**
* This delegates to the <Event_Channel> to do the actual
* processing. Typically, it forwards the <event> to its
* appropriate Consumer.
*/
int process (ACE_Message_Block *event);
/// Keep track of event fragment to handle non-blocking recv's from
/// Suppliers.
ACE_Message_Block *msg_frag_;
};
/**
* @class Consumer_Handler
*
* @brief Handles transmission of events to Consumers.
*
* Performs queueing and error checking. Intended to run
* reactively, i.e., in one thread of control using a Reactor
* for demuxing and dispatching. Also uses a Reactor to handle
* flow controlled output connections.
*/
class Consumer_Handler : public Connection_Handler
{
public:
Consumer_Handler (const Connection_Config_Info &);
/// Send an event to a Consumer (may be queued if necessary).
virtual int put (ACE_Message_Block *event,
ACE_Time_Value * = 0);
protected:
/// Finish sending event when flow control conditions abate.
virtual int handle_output (ACE_HANDLE);
/// Perform a non-blocking put().
int nonblk_put (ACE_Message_Block *mb);
/// Send an event to a Consumer.
virtual ssize_t send (ACE_Message_Block *);
/// Receive and process shutdowns from a Consumer.
virtual int handle_input (ACE_HANDLE);
};
/**
* @class Thr_Consumer_Handler
*
* @brief Runs each <Consumer_Handler> in a separate thread.
*/
class Thr_Consumer_Handler : public Consumer_Handler
{
public:
Thr_Consumer_Handler (const Connection_Config_Info &);
/// Initialize the threaded Consumer_Handler object and spawn a new
/// thread.
virtual int open (void *);
/// Send a message to a peer.
virtual int put (ACE_Message_Block *, ACE_Time_Value * = 0);
protected:
/// Called when Peer shutdown unexpectedly.
virtual int handle_input (ACE_HANDLE);
/// Transmit peer messages.
virtual int svc (void);
/**
* When thread started, connection become blocked, so no need to use
* handle_close to reinitiate the connection_handler, so should
* override this function to justify if controlling is in thread or
* not. If yes, handle_close do nothing, otherwise, it call parent
* handle_close().
*/
virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,
ACE_Reactor_Mask = ACE_Event_Handler::ALL_EVENTS_MASK);
private:
/// If the controlling is in thread's svc() or not.
int in_thread_;
};
/**
* @class Thr_Supplier_Handler
*
* @brief Runs each <Supplier_Handler> in a separate thread.
*/
class Thr_Supplier_Handler : public Supplier_Handler
{
public:
Thr_Supplier_Handler (const Connection_Config_Info &pci);
/// Initialize the object and spawn a new thread.
virtual int open (void *);
protected:
/**
* When thread started, connection become blocked, so no need to use
* handle_close to reinitiate the connection_handler, so should
* override this function to justify if controlling is in thread or
* not. If yes, handle_close do nothing, otherwise, it call parent
* handle_close().
*/
virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,
ACE_Reactor_Mask = ACE_Event_Handler::ALL_EVENTS_MASK);
/// Transmit peer messages.
virtual int svc (void);
private:
/// If the controlling is in thread's svc() or not.
int in_thread_;
};
#endif /* CONCRETE_CONNECTION_HANDLER */
|
