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
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
|
// **********************************************************************
//
// Copyright (c) 2003-2014 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************
#ifndef ICE_OUTGOING_ASYNC_H
#define ICE_OUTGOING_ASYNC_H
#include <IceUtil/Timer.h>
#include <Ice/OutgoingAsyncF.h>
#include <Ice/AsyncResult.h>
#include <Ice/CommunicatorF.h>
#include <Ice/ConnectionIF.h>
#include <Ice/ObjectAdapterF.h>
namespace IceInternal
{
class RetryException;
class CollocatedRequestHandler;
//
// Base class for handling asynchronous invocations. This class is
// responsible for the handling of the output stream and the child
// invocation observer.
//
class ICE_API OutgoingAsyncBase : public Ice::AsyncResult
{
public:
//
// Those methods must be overriden if the invocation is sent
// through a request handler.
//
virtual AsyncStatus send(const Ice::ConnectionIPtr&, bool, bool) { assert(false); return AsyncStatusQueued; }
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*) { assert(false); return AsyncStatusQueued; }
virtual bool sent();
virtual bool completed(const Ice::Exception&);
virtual void retryException(const Ice::Exception&);
// Those methods are public when called from an OutgoingAsyncBase reference.
using Ice::AsyncResult::cancelable;
using Ice::AsyncResult::invokeSent;
using Ice::AsyncResult::invokeSentAsync;
using Ice::AsyncResult::invokeCompleted;
using Ice::AsyncResult::invokeCompletedAsync;
void attachRemoteObserver(const Ice::ConnectionInfoPtr& c, const Ice::EndpointPtr& endpt, Ice::Int requestId)
{
const Ice::Int size = static_cast<Ice::Int>(_os.b.size() - headerSize - 4);
_childObserver.attach(getObserver().getRemoteObserver(c, endpt, requestId, size));
}
void attachCollocatedObserver(const Ice::ObjectAdapterPtr& adapter, Ice::Int requestId)
{
const Ice::Int size = static_cast<Ice::Int>(_os.b.size() - headerSize - 4);
_childObserver.attach(getObserver().getCollocatedObserver(adapter, requestId, size));
}
BasicStream* getOs()
{
return &_os;
}
protected:
OutgoingAsyncBase(const Ice::CommunicatorPtr&, const InstancePtr&, const std::string&, const CallbackBasePtr&,
const Ice::LocalObjectPtr&);
bool sent(bool);
bool finished(const Ice::Exception&);
ObserverHelperT<Ice::Instrumentation::ChildInvocationObserver> _childObserver;
BasicStream _os;
};
//
// Base class for proxy based invocations. This class handles the
// retry for proxy invocations. It also ensures the child observer is
// correct notified of failures and make sure the retry task is
// correctly canceled when the invocation completes.
//
class ICE_API ProxyOutgoingAsyncBase : public OutgoingAsyncBase, protected IceUtil::TimerTask
{
public:
virtual Ice::ObjectPrx getProxy() const;
using OutgoingAsyncBase::sent;
virtual bool completed(const Ice::Exception&);
virtual void retryException(const Ice::Exception&);
virtual void cancelable(const CancellationHandlerPtr&);
void retry();
void abort(const Ice::Exception&);
protected:
ProxyOutgoingAsyncBase(const Ice::ObjectPrx&, const std::string&, const CallbackBasePtr&,
const Ice::LocalObjectPtr&);
void invokeImpl(bool);
bool sent(bool);
bool finished(const Ice::Exception&);
bool finished(bool);
virtual void handleRetryException(const Ice::Exception&);
virtual int handleException(const Ice::Exception&);
virtual void runTimerTask();
const Ice::ObjectPrx _proxy;
RequestHandlerPtr _handler;
Ice::OperationMode _mode;
private:
int _cnt;
bool _sent;
};
//
// Class for handling Slice operation invocations
//
class ICE_API OutgoingAsync : public ProxyOutgoingAsyncBase
{
public:
OutgoingAsync(const Ice::ObjectPrx&, const std::string&, const CallbackBasePtr&, const Ice::LocalObjectPtr&);
void prepare(const std::string&, Ice::OperationMode, const Ice::Context*);
virtual bool sent();
virtual AsyncStatus send(const Ice::ConnectionIPtr&, bool, bool);
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*);
void abort(const Ice::Exception&);
void invoke();
using ProxyOutgoingAsyncBase::completed;
bool completed();
BasicStream* startWriteParams(Ice::FormatType format)
{
_os.startWriteEncaps(_encoding, format);
return &_os;
}
void endWriteParams()
{
_os.endWriteEncaps();
}
void writeEmptyParams()
{
_os.writeEmptyEncaps(_encoding);
}
void writeParamEncaps(const ::Ice::Byte* encaps, ::Ice::Int size)
{
if(size == 0)
{
_os.writeEmptyEncaps(_encoding);
}
else
{
_os.writeEncaps(encaps, size);
}
}
BasicStream* getIs()
{
return &_is;
}
private:
const Ice::EncodingVersion _encoding;
};
//
// Class for handling the proxy's begin_ice_flushBatchRequest request.
//
class ICE_API ProxyFlushBatch : public ProxyOutgoingAsyncBase
{
public:
ProxyFlushBatch(const Ice::ObjectPrx&, const std::string&, const CallbackBasePtr&, const Ice::LocalObjectPtr&);
virtual AsyncStatus send(const Ice::ConnectionIPtr&, bool, bool);
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*);
void invoke();
private:
virtual void handleRetryException(const Ice::Exception&);
virtual int handleException(const Ice::Exception&);
};
typedef IceUtil::Handle<ProxyFlushBatch> ProxyFlushBatchPtr;
//
// Class for handling the proxy's begin_ice_getConnection request.
//
class ICE_API ProxyGetConnection : public ProxyOutgoingAsyncBase
{
public:
ProxyGetConnection(const Ice::ObjectPrx&, const std::string&, const CallbackBasePtr&, const Ice::LocalObjectPtr&);
virtual AsyncStatus send(const Ice::ConnectionIPtr&, bool, bool);
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*);
void invoke();
};
typedef IceUtil::Handle<ProxyGetConnection> ProxyGetConnectionPtr;
//
// Class for handling Ice::Connection::begin_flushBatchRequests
//
class ICE_API ConnectionFlushBatch : public OutgoingAsyncBase
{
public:
ConnectionFlushBatch(const Ice::ConnectionIPtr&, const Ice::CommunicatorPtr&, const InstancePtr&,
const std::string&, const CallbackBasePtr&, const Ice::LocalObjectPtr&);
virtual Ice::ConnectionPtr getConnection() const;
void invoke();
private:
const Ice::ConnectionIPtr _connection;
};
typedef IceUtil::Handle<ConnectionFlushBatch> ConnectionFlushBatchPtr;
//
// Class for handling Ice::Communicator::begin_flushBatchRequests
//
class ICE_API CommunicatorFlushBatch : public Ice::AsyncResult
{
public:
CommunicatorFlushBatch(const Ice::CommunicatorPtr&, const InstancePtr&, const std::string&,
const CallbackBasePtr&, const Ice::LocalObjectPtr&);
void flushConnection(const Ice::ConnectionIPtr&);
void ready();
private:
void check(bool);
int _useCount;
};
}
#endif
|