bartoszpop commented on code in PR #25908:
URL: https://github.com/apache/camel/pull/25908#discussion_r3896141763
##########
components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java:
##########
@@ -129,108 +136,175 @@ public SubscriptionHelper(final SalesforceComponent
component) {
}
private MessageListener createHandshakeListener() {
- return (channel, message) ->
component.getHttpClient().getWorkerPool().execute(() -> {
- LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message);
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Handshake failure: {}", message);
+ handshakeError = (String)
message.get(ERROR_FIELD);
+ handshakeException = getFailure(message);
+ if (handshakeError != null) {
+ if (handshakeError.startsWith("403::")) {
+ String failureReason =
getFailureReason(message);
+ if
(AUTHENTICATION_INVALID.equals(failureReason)) {
+ LOG.debug(
+ "attempting login due to
handshake error: 403 -> 401::Authentication invalid");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ }
+ }
+ // failed, so keep trying with backoff
+ final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
+ if (backoff > maxBackoff) {
+ LOG.error("Handshake retry aborted after
exceeding {} msecs backoff", maxBackoff);
+ } else {
+ LOG.debug("Pausing for {} msecs before
handshake retry", backoff);
+ if (backoff > 0) {
+ Tasks.backgroundTask()
+
.withBudget(Budgets.iterationTimeBudget()
+ .withMaxIterations(1)
+
.withInitialDelay(Duration.ofMillis(backoff))
+
.withInterval(Duration.ofMillis(1))
+
.withUnlimitedDuration()
+ .build())
+
.withScheduledExecutor(taskExecutor)
+
.withName("SalesforceHandshakeRetryDelay")
+ .build()
+
.run(component.getCamelContext(), () -> true);
+ }
+ client.handshake();
+ }
+ } else if (!channelToConsumers.isEmpty()) {
+ channelsLock.lock();
+ try {
+ channelsToSubscribe.clear();
+
channelsToSubscribe.addAll(channelToConsumers.keySet());
+ } finally {
+ channelsLock.unlock();
+ }
+ LOG.info("Handshake successful. Channels to
subscribe: {}", channelsToSubscribe);
+ }
+ });
+ }
- if (!message.isSuccessful()) {
- LOG.warn("Handshake failure: {}", message);
- handshakeError = (String) message.get(ERROR_FIELD);
- handshakeException = getFailure(message);
- if (handshakeError != null) {
- if (handshakeError.startsWith("403::")) {
- String failureReason = getFailureReason(message);
- if (AUTHENTICATION_INVALID.equals(failureReason)) {
- LOG.debug(
- "attempting login due to handshake error:
403 -> 401::Authentication invalid");
-
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
- }
- }
- }
- // failed, so keep trying with backoff
- final long backoff =
handshakeBackoff.getAndAdd(backoffIncrement);
- if (backoff > maxBackoff) {
- LOG.error("Handshake retry aborted after exceeding {}
msecs backoff", maxBackoff);
- } else {
- LOG.debug("Pausing for {} msecs before handshake retry",
backoff);
- if (backoff > 0) {
- Tasks.backgroundTask()
- .withBudget(Budgets.iterationTimeBudget()
- .withMaxIterations(1)
-
.withInitialDelay(Duration.ofMillis(backoff))
- .withInterval(Duration.ofMillis(1))
- .withUnlimitedDuration()
- .build())
- .withScheduledExecutor(taskExecutor)
- .withName("SalesforceHandshakeRetryDelay")
- .build()
- .run(component.getCamelContext(), () -> true);
- }
- client.handshake();
- }
- } else if (!channelToConsumers.isEmpty()) {
- channelsLock.lock();
- try {
- channelsToSubscribe.clear();
- channelsToSubscribe.addAll(channelToConsumers.keySet());
- } finally {
- channelsLock.unlock();
+ private MessageListener createConnectionListener() {
+ return (channel, message) -> component
+ .getHttpClient()
+ .getWorkerPool()
+ .execute(
+ () -> {
+ LOG.debug("[CHANNEL:META_CONNECT]: {}", message);
+ String reconnectAdvice = message.getAdvice() !=
null
+ ? (String)
message.getAdvice().get("reconnect")
+ : null;
+
+ if (!message.isSuccessful()) {
+ LOG.warn("Connect failure: {}", message);
+ connectError = (String)
message.get(ERROR_FIELD);
+ connectException = getFailure(message);
+
+ if (connectError != null &&
connectError.equals(AUTHENTICATION_INVALID)) {
+ LOG.debug("connectError: {}",
connectError);
+ LOG.debug("Attempting login...");
+
session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff);
+ }
+ // Per Bayeux spec: handshake on null advice,
"none", "handshake", or any non-"retry" value.
+ // When advice is "retry", the CometD client
handles reconnection automatically.
+ if (reconnectAdvice == null ||
!"retry".equals(reconnectAdvice)) {
+ LOG.debug("Reconnect advice [{}] on failed
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else if (isTemporaryError(message)) {
+ LOG.debug("Initiating handshake after
temporary error: {}", message);
+ client.handshake();
+ }
+ } else if (reconnectAdvice != null &&
!"retry".equals(reconnectAdvice)) {
+ LOG.warn("Reconnect advice [{}] on successful
connect, initiating handshake", reconnectAdvice);
+ client.handshake();
+ } else {
+ Set<String> toSubscribe = null;
+ channelsLock.lock();
+ try {
+ if (!channelsToSubscribe.isEmpty()) {
+ toSubscribe = new
HashSet<>(channelsToSubscribe);
+ channelsToSubscribe.clear();
+ }
+ } finally {
+ channelsLock.unlock();
+ }
+ if (toSubscribe != null) {
+ LOG.info("Subscribing to channels: {}",
toSubscribe);
+ for (var channelName : toSubscribe) {
+ var consumers =
channelToConsumers.getOrDefault(channelName, emptySet());
+ for (var consumer : consumers) {
+ subscribe(consumer);
+ }
+ }
+ }
+ }
+ });
+ }
+
+ private MessageListener createDisconnectListener() {
+ return (channel, message) -> {
+ LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message);
+
+ if (isStoppingOrStopped()) {
+ LOG.debug("Ignoring disconnect message while stopping");
+ return;
+ }
+ if (!reconnecting.compareAndSet(false, true)) {
+ LOG.debug("Reconnect already in progress");
+ return;
+ }
+
+ final ScheduledExecutorService executor = taskExecutor;
+ if (executor == null) {
+ reconnecting.set(false);
+ return;
+ }
+
+ try {
+ executor.execute(this::reconnectAfterDisconnect);
+ } catch (RejectedExecutionException e) {
+ reconnecting.set(false);
+ if (!isStoppingOrStopped()) {
+ LOG.warn("Unable to schedule reconnect after server
disconnect", e);
}
- LOG.info("Handshake successful. Channels to subscribe: {}",
channelsToSubscribe);
}
- });
+ };
}
- private MessageListener createConnectionListener() {
- return (channel, message) ->
component.getHttpClient().getWorkerPool().execute(() -> {
- LOG.debug("[CHANNEL:META_CONNECT]: {}", message);
- String reconnectAdvice = message.getAdvice() != null
- ? (String) message.getAdvice().get("reconnect")
- : null;
+ private void reconnectAfterDisconnect() {
+ final BayeuxClient disconnectedClient = client;
+ try {
+ if (disconnectedClient == null || isStoppingOrStopped()) {
+ return;
+ }
- if (!message.isSuccessful()) {
- LOG.warn("Connect failure: {}", message);
- connectError = (String) message.get(ERROR_FIELD);
- connectException = getFailure(message);
-
- if (connectError != null &&
connectError.equals(AUTHENTICATION_INVALID)) {
- LOG.debug("connectError: {}", connectError);
- LOG.debug("Attempting login...");
- session.attemptLoginUntilSuccessful(backoffIncrement,
maxBackoff);
- }
- // Per Bayeux spec: handshake on null advice, "none",
"handshake", or any non-"retry" value.
- // When advice is "retry", the CometD client handles
reconnection automatically.
- if (reconnectAdvice == null ||
!"retry".equals(reconnectAdvice)) {
- LOG.debug("Reconnect advice [{}] on failed connect,
initiating handshake", reconnectAdvice);
- client.handshake();
- } else if (isTemporaryError(message)) {
- LOG.debug("Initiating handshake after temporary error:
{}", message);
- client.handshake();
- }
- } else if (reconnectAdvice != null &&
!"retry".equals(reconnectAdvice)) {
- LOG.warn("Reconnect advice [{}] on successful connect,
initiating handshake", reconnectAdvice);
- client.handshake();
- } else {
- Set<String> toSubscribe = null;
- channelsLock.lock();
- try {
- if (!channelsToSubscribe.isEmpty()) {
- toSubscribe = new HashSet<>(channelsToSubscribe);
- channelsToSubscribe.clear();
- }
- } finally {
- channelsLock.unlock();
- }
- if (toSubscribe != null) {
- LOG.info("Subscribing to channels: {}", toSubscribe);
- for (var channelName : toSubscribe) {
- var consumers =
channelToConsumers.getOrDefault(channelName, emptySet());
- for (var consumer : consumers) {
- subscribe(consumer);
- }
- }
+ final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC,
SECONDS);
+ if (!disconnectedClient.waitFor(waitMs,
BayeuxClient.State.DISCONNECTED)) {
Review Comment:
That's a valid point. I changed it to invoke reconnectAfterDisconnect in the
http worker thread. Also, I reduced the disconnect timeout to 10 sec
--
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]
