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
|
// **********************************************************************
//
// Copyright (c) 2003-2017 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:
virtual bool sent();
virtual bool completed(const Ice::Exception&);
virtual bool completed();
// 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;
}
virtual BasicStream* getIs();
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 AsyncStatus invokeRemote(const Ice::ConnectionIPtr&, bool, bool) = 0;
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*) = 0;
virtual Ice::ObjectPrx getProxy() const;
using OutgoingAsyncBase::sent;
virtual bool completed(const Ice::Exception&);
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);
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 invokeRemote(const Ice::ConnectionIPtr&, bool, bool);
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*);
using ProxyOutgoingAsyncBase::completed;
virtual bool completed();
void abort(const Ice::Exception&);
void invoke();
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);
}
}
virtual BasicStream* getIs()
{
return &_is;
}
private:
const Ice::EncodingVersion _encoding;
};
//
// Class for handling the proxy's begin_ice_flushBatchRequest request.
//
class ICE_API ProxyFlushBatchAsync : public ProxyOutgoingAsyncBase
{
public:
ProxyFlushBatchAsync(const Ice::ObjectPrx&, const std::string&, const CallbackBasePtr&, const Ice::LocalObjectPtr&);
virtual AsyncStatus invokeRemote(const Ice::ConnectionIPtr&, bool, bool);
virtual AsyncStatus invokeCollocated(CollocatedRequestHandler*);
void invoke();
private:
int _batchRequestNum;
};
typedef IceUtil::Handle<ProxyFlushBatchAsync> ProxyFlushBatchAsyncPtr;
//
// 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 invokeRemote(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 ConnectionFlushBatchAsync : public OutgoingAsyncBase
{
public:
ConnectionFlushBatchAsync(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<ConnectionFlushBatchAsync> ConnectionFlushBatchAsyncPtr;
//
// Class for handling Ice::Communicator::begin_flushBatchRequests
//
class ICE_API CommunicatorFlushBatchAsync : public Ice::AsyncResult
{
public:
CommunicatorFlushBatchAsync(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
|
