Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Incorporated review comments. 

- Added description on what session maintenance does.
- Added calling deregister when initial connection attempt fails so that a processor can retry connecting at next onTrigger.

This closes #1597

Signed-off-by: Jeremy Dyer <jeremydyer@apache.org>
This commit is contained in:
Koji Kawamura 2017-05-02 17:12:34 +09:00 committed by Jeremy Dyer
parent 0a014b471b
commit cca5c42095
2 changed files with 16 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-processors/src/main/java/org/apache/nifi/processors/websocket/AbstractWebSocketGatewayProcessor.java
View File

@ -147,6 +147,10 @@ public abstract class AbstractWebSocketGatewayProcessor extends AbstractSessionF
@OnStopped
public void onStopped(final ProcessContext context) throws IOException {
deregister();
}
private void deregister() {
if (webSocketService == null) {
return;
}
@ -170,11 +174,14 @@ public abstract class AbstractWebSocketGatewayProcessor extends AbstractSessionF
try {
registerProcessorToService(context, webSocketService -> onWebSocketServiceReady(webSocketService));
} catch (IOException|WebSocketConfigurationException e) {
// Deregister processor if it failed so that it can retry next onTrigger.
deregister();
context.yield();
throw new ProcessException("Failed to register processor to WebSocket service due to: " + e, e);
}
}
context.yield();//nothing really to do here since threading managed by smtp server sessions
context.yield();//nothing really to do here since handling WebSocket messages is done at ControllerService.
}

10
nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/main/java/org/apache/nifi/websocket/jetty/JettyWebSocketClient.java
View File

@ -90,7 +90,12 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
public static final PropertyDescriptor SESSION_MAINTENANCE_INTERVAL = new PropertyDescriptor.Builder()
.name("session-maintenance-interval")
.displayName("Session Maintenance Interval")
.description("The interval between session maintenance activities.")
.description("The interval between session maintenance activities." +
" A WebSocket session established with a WebSocket server can be terminated due to different reasons" +
" including restarting the WebSocket server or timing out inactive sessions." +
" This session maintenance activity is periodically executed in order to reconnect those lost sessions," +
" so that a WebSocket client can reuse the same session id transparently after it reconnects successfully. " +
" The maintenance activity is executed until corresponding processors or this controller service is stopped.")
.required(true)
.expressionLanguageSupported(true)
.addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
@ -238,10 +243,11 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
}
final String sessionId = activeSessions.get(clientId);
// If this session is stil alive, do nothing.
// If this session is still alive, do nothing.
if (!router.containsSession(sessionId)) {
// This session is no longer active, reconnect it.
// If it fails, the sessionId will remain in activeSessions, and retries later.
// This reconnect attempt is continued until user explicitly stops a processor or this controller service.
connect(clientId, sessionId);
}
}