gnodet commented on code in PR #28537:
URL: https://github.com/apache/camel/pull/28537#discussion_r4387141092
##########
components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java:
##########
@@ -124,27 +135,110 @@ protected void doStart() throws Exception {
if (asyncTransport == null) {
initializeAsyncTransport();
LOG.info("Getting asynchronous client implementation");
+ asyncClientManager = new TAsyncClientManager();
thriftClient =
ThriftUtils.constructAsyncClientInstance(endpoint.getServicePackage(),
endpoint.getServiceName(),
asyncTransport,
- configuration.getExchangeProtocol(),
endpoint.getCamelContext());
+ configuration.getExchangeProtocol(),
endpoint.getCamelContext(), asyncClientManager);
}
}
}
@Override
protected void doStop() throws Exception {
- if (syncTransport != null) {
- LOG.debug("Terminating synchronous transport the remote Thrift
server");
- syncTransport.close();
- syncTransport = null;
- } else if (asyncTransport != null) {
- LOG.debug("Terminating asynchronous transport the remote Thrift
server");
- asyncTransport.close();
- asyncTransport = null;
+ try {
+ if (syncTransport != null) {
+ LOG.debug("Terminating synchronous transport the remote Thrift
server");
+ syncTransport.close();
+ syncTransport = null;
+ } else if (asyncTransport != null) {
+ LOG.debug("Terminating asynchronous transport the remote
Thrift server");
+ asyncTransport.close();
+ asyncTransport = null;
+ }
+ } finally {
+ if (asyncClientManager != null) {
+ // stops the selector thread and closes the selector of the
async client
+ asyncClientManager.stop();
+ asyncClientManager = null;
+ }
+ // after the stop, so that a call registered meanwhile is failed
too
+ failPendingCallbacks();
}
super.doStop();
}
+ /**
+ * Fails the exchanges of the asynchronous calls that have not completed:
the stopped selector thread does not
+ * complete them. A call that completes meanwhile is completed once, and a
continuation that throws does not skip
+ * the other calls.
+ */
+ private void failPendingCallbacks() {
+ for (PendingCallback pending : pendingCallbacks) {
+ try {
+ pending.stopped();
+ } catch (Exception e) {
+ LOG.warn("Error completing the exchange of a pending call of
method {} when stopping the Thrift producer",
+ configuration.getMethod(), e);
+ }
+ }
+ }
+
+ /**
+ * The number of asynchronous calls that have not completed yet (for
tests).
+ */
+ int getPendingCallbackCount() {
+ return pendingCallbacks.size();
+ }
+
+ /**
+ * Completes the exchange of an asynchronous call once: with the response
or error of the call, or with an error
+ * when the producer stops first.
+ */
+ private final class PendingCallback extends AsyncClientMethodCallback {
+ private final Exchange exchange;
+ private final AtomicBoolean done = new AtomicBoolean();
+
+ PendingCallback(Exchange exchange, AsyncCallback callback) {
+ super(exchange, callback);
+ this.exchange = exchange;
+ }
+
+ @Override
+ public void onComplete(Object response) {
+ if (finish()) {
+ super.onComplete(response);
+ }
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ if (finish()) {
+ super.onError(exception);
+ }
+ }
+
+ void stopped() {
+ if (finish()) {
+ super.onError(new CamelExchangeException(
+ "The Thrift producer was stopped before the call of
method " + configuration.getMethod()
+ + " completed",
+ exchange));
+ }
+ }
+
+ void discard() {
Review Comment:
Return `finish()`'s boolean so callers can react:
```java
boolean discard() {
return finish();
}
```
##########
components/camel-thrift/src/main/java/org/apache/camel/component/thrift/ThriftProducer.java:
##########
@@ -60,10 +68,13 @@ public ThriftProducer(ThriftEndpoint endpoint,
ThriftConfiguration configuration
public boolean process(Exchange exchange, AsyncCallback callback) {
Message message = exchange.getIn();
+ PendingCallback pending = new PendingCallback(exchange, callback);
+ pendingCallbacks.add(pending);
try {
- ThriftUtils.invokeAsyncMethod(thriftClient,
configuration.getMethod(), message.getBody(),
- new AsyncClientMethodCallback(exchange, callback));
+ ThriftUtils.invokeAsyncMethod(thriftClient,
configuration.getMethod(), message.getBody(), pending);
} catch (Exception e) {
+ // the call was not made
+ pending.discard();
Review Comment:
`discard()` returns `void`, so the race result from `finish()` is lost. If
`doStop()` concurrently calls `stopped()` and wins the CAS here first
(completing the exchange via `callback.done(false)` in `super.onError(...)`),
the catch block continues unconditionally to `exchange.setException(e)` and
`callback.done(true)` — a double-completion.
Guard it:
```java
if (pending.discard()) {
if (e.getCause() instanceof TException) {
exchange.setException(e.getCause());
} else {
exchange.setException(e);
}
callback.done(true);
}
return true;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
