blob: 2c28116da1fad5eae3984db791bbd004811cd246 (
plain)
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
|
package org.apache.qpid.messaging.util;
import org.apache.qpid.messaging.ConnectionException;
import org.apache.qpid.messaging.Message;
import org.apache.qpid.messaging.MessagingException;
import org.apache.qpid.messaging.Receiver;
import org.apache.qpid.messaging.ReceiverException;
import org.apache.qpid.messaging.Sender;
import org.apache.qpid.messaging.SenderException;
import org.apache.qpid.messaging.Session;
import org.apache.qpid.messaging.SessionException;
import org.apache.qpid.messaging.ext.SenderExt;
import org.apache.qpid.messaging.ext.SessionExt;
import org.apache.qpid.messaging.util.ReceiverManagementDecorator.ReceiverState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* A Decorator that adds basic housekeeping tasks to a Sender.
* This class adds,
* 1. State management.
* 2. Exception handling.
*
*/
public class SenderManagementDecorator implements SenderExt
{
private static Logger _logger = LoggerFactory.getLogger(SenderManagementDecorator.class);
public enum SenderState {OPENED, CLOSED, ERROR};
private Sender _delegate;
private SenderState _state = SenderState.OPENED;
private SessionExt _ssn;
private final Object _connectionLock; // global per connection lock
public SenderManagementDecorator(SessionExt ssn, Sender delegate)
{
_ssn = ssn;
_delegate = delegate;
_connectionLock = ssn.getConnectionExt().getConnectionLock();
}
@Override
public void send(Message message, boolean sync) throws MessagingException
{
checkClosedAndThrowException();
try
{
_delegate.send(message, sync);
}
catch (ConnectionException e)
{
throw handleConnectionException(e);
}
catch (SessionException e)
{
throw handleSessionException(e);
}
}
@Override
public void close() throws MessagingException
{
checkClosedAndThrowException("Sender is already closed");
synchronized (_connectionLock)
{
_state = SenderState.CLOSED;
_delegate.close();
}
}
@Override
public void setCapacity(int capacity) throws MessagingException
{
checkClosedAndThrowException();
try
{
_delegate.setCapacity(capacity);
}
catch (ConnectionException e)
{
throw handleConnectionException(e);
}
catch (SessionException e)
{
throw handleSessionException(e);
}
}
@Override
public int getCapacity() throws MessagingException
{
checkClosedAndThrowException();
try
{
return _delegate.getCapacity();
}
catch (ConnectionException e)
{
throw handleConnectionException(e);
}
catch (SessionException e)
{
throw handleSessionException(e);
}
}
@Override
public int getAvailable() throws MessagingException
{
checkClosedAndThrowException();
try
{
return _delegate.getAvailable();
}
catch (ConnectionException e)
{
throw handleConnectionException(e);
}
catch (SessionException e)
{
throw handleSessionException(e);
}
}
@Override
public int getUnsettled() throws MessagingException
{
checkClosedAndThrowException();
try
{
return _delegate.getUnsettled();
}
catch (ConnectionException e)
{
throw handleConnectionException(e);
}
catch (SessionException e)
{
throw handleSessionException(e);
}
}
@Override
public boolean isClosed() throws MessagingException
{
return _state == SenderState.CLOSED;
}
@Override
public String getName() throws MessagingException
{
checkClosedAndThrowException();
return _delegate.getName();
}
@Override
public Session getSession() throws MessagingException
{
checkClosedAndThrowException();
_ssn.checkError();
return _ssn;
}
@Override
public void recreate() throws MessagingException
{
// TODO Auto-generated method stub
}
private void checkClosedAndThrowException() throws ReceiverException
{
checkClosedAndThrowException("Receiver is closed. You cannot invoke methods on a closed receiver");
}
private void checkClosedAndThrowException(String closedMessage) throws ReceiverException
{
switch (_state)
{
case ERROR:
throw new ReceiverException("Receiver is in a temporary error state. The session may or may not recover from this");
case CLOSED:
throw new ReceiverException(closedMessage);
}
}
/**
* A ConnectionException will cause the Session/Sender to go into a temporary error state,
* which prevents it from being used further.
* From there the Session and Sender can be moved into OPENED (if failover works) or
* CLOSED if there is no failover or if failover has failed.
* @param e
* @throws MessagingException
*/
private SenderException handleConnectionException(ConnectionException e)
{
synchronized (_connectionLock)
{
_state = SenderState.ERROR;
_ssn.exception(e); // This might trigger failover in a layer above.
if (_state == SenderState.CLOSED)
{
// The connection has instructed the session and it's child objects to be closed.
// Either there was no failover, or failover has failed.
return new SenderException("Sender is closed due to connection error",e);
}
else
{
// Asking the application or the Parent handler to retry the operation.
// The Session and Receiver should be in OPENED state at this time.
return new SenderException("Sender was in a temporary error state due to connection error." +
"Plase retry your operation",e);
}
}
}
/**
* Session Exceptions will generally invalidate the Session.
* TODO this needs to be revisited again.
* A new session will need to be created in that case.
* @param e
* @throws MessagingException
*/
private SenderException handleSessionException(SessionException e)
{
synchronized (_connectionLock)
{
// This should close all senders (including this) and receivers.
_ssn.exception(e);
}
return new SenderException("Session has been closed",e);
}
}
|
