[
https://issues.apache.org/jira/browse/CAMEL-12005?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16288962#comment-16288962
]
ASF GitHub Bot commented on CAMEL-12005:
----------------------------------------
davsclaus commented on a change in pull request #2144: CAMEL-12005: Add
websocket support to camel-undertow
URL: https://github.com/apache/camel/pull/2144#discussion_r156601727
##########
File path:
components/camel-undertow/src/main/java/org/apache/camel/component/undertow/UndertowProducer.java
##########
@@ -77,81 +83,135 @@ public UndertowEndpoint getEndpoint() {
return endpoint;
}
+ boolean isSendToAll(Message in) {
+ // header may be null; have to be careful here (and fallback to use
sendToAll option configured from endpoint)
+ Boolean value = in.getHeader(UndertowConstants.SEND_TO_ALL,
endpoint.getSendToAll(), Boolean.class);
+ return value == null ? false : value;
+ }
+
@Override
public boolean process(final Exchange camelExchange, final AsyncCallback
callback) {
- final URI uri;
- final HttpString method;
- try {
- final String exchangeUri = UndertowHelper.createURL(camelExchange,
getEndpoint());
- uri = UndertowHelper.createURI(camelExchange, exchangeUri,
getEndpoint());
- method = UndertowHelper.createMethod(camelExchange, endpoint,
camelExchange.getIn().getBody() != null);
- } catch (final URISyntaxException e) {
- camelExchange.setException(e);
- callback.done(true);
- return true;
- }
-
- final String pathAndQuery = URISupport.pathAndQueryOf(uri);
-
- final UndertowHttpBinding undertowHttpBinding =
endpoint.getUndertowHttpBinding();
-
- final CookieHandler cookieHandler = endpoint.getCookieHandler();
- final Map<String, List<String>> cookieHeaders;
- if (cookieHandler != null) {
+ if (endpoint.isWebSocket()) {
+ return processWebSocket(camelExchange, callback);
+ } else {
+ /* not a WebSocket */
+ final URI uri;
+ final HttpString method;
try {
- cookieHeaders = cookieHandler.loadCookies(camelExchange, uri);
- } catch (final IOException e) {
+ final String exchangeUri =
UndertowHelper.createURL(camelExchange, getEndpoint());
+ uri = UndertowHelper.createURI(camelExchange, exchangeUri,
getEndpoint());
+ method = UndertowHelper.createMethod(camelExchange, endpoint,
camelExchange.getIn().getBody() != null);
+ } catch (final URISyntaxException e) {
camelExchange.setException(e);
callback.done(true);
return true;
}
- } else {
- cookieHeaders = Collections.emptyMap();
- }
- final ClientRequest request = new ClientRequest();
- request.setMethod(method);
- request.setPath(pathAndQuery);
+ final String pathAndQuery = URISupport.pathAndQueryOf(uri);
+
+ final UndertowHttpBinding undertowHttpBinding =
endpoint.getUndertowHttpBinding();
+
+ final CookieHandler cookieHandler = endpoint.getCookieHandler();
+ final Map<String, List<String>> cookieHeaders;
+ if (cookieHandler != null) {
+ try {
+ cookieHeaders = cookieHandler.loadCookies(camelExchange,
uri);
+ } catch (final IOException e) {
+ camelExchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+ } else {
+ cookieHeaders = Collections.emptyMap();
+ }
- final HeaderMap requestHeaders = request.getRequestHeaders();
+ final ClientRequest request = new ClientRequest();
+ request.setMethod(method);
+ request.setPath(pathAndQuery);
- // Set the Host header
- final Message message = camelExchange.getIn();
- final String host = message.getHeader(Headers.HOST_STRING,
String.class);
- requestHeaders.put(Headers.HOST,
Optional.ofNullable(host).orElseGet(() -> uri.getAuthority()));
+ final HeaderMap requestHeaders = request.getRequestHeaders();
- final Object body = undertowHttpBinding.toHttpRequest(request,
camelExchange.getIn());
+ // Set the Host header
+ final Message message = camelExchange.getIn();
+ final String host = message.getHeader(Headers.HOST_STRING,
String.class);
+ requestHeaders.put(Headers.HOST,
Optional.ofNullable(host).orElseGet(() -> uri.getAuthority()));
- final TypeConverter tc = endpoint.getCamelContext().getTypeConverter();
- final ByteBuffer bodyAsByte = tc.tryConvertTo(ByteBuffer.class, body);
+ final Object body = undertowHttpBinding.toHttpRequest(request,
camelExchange.getIn());
- // As tryConvertTo is used to convert the body, we should do null check
- // or the call bodyAsByte.remaining() may throw an NPE
- if (body != null && bodyAsByte != null) {
- requestHeaders.put(Headers.CONTENT_LENGTH, bodyAsByte.remaining());
- }
+ final TypeConverter tc =
endpoint.getCamelContext().getTypeConverter();
+ final ByteBuffer bodyAsByte = tc.tryConvertTo(ByteBuffer.class,
body);
- for (final Map.Entry<String, List<String>> entry :
cookieHeaders.entrySet()) {
- requestHeaders.putAll(HttpString.tryFromString(entry.getKey()),
entry.getValue());
- }
+ // As tryConvertTo is used to convert the body, we should do null
check
+ // or the call bodyAsByte.remaining() may throw an NPE
+ if (body != null && bodyAsByte != null) {
+ requestHeaders.put(Headers.CONTENT_LENGTH,
bodyAsByte.remaining());
+ }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Executing http {} method: {}", method, pathAndQuery);
+ for (final Map.Entry<String, List<String>> entry :
cookieHeaders.entrySet()) {
+
requestHeaders.putAll(HttpString.tryFromString(entry.getKey()),
entry.getValue());
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Executing http {} method: {}", method,
pathAndQuery);
+ }
+
+ final UndertowClientCallback clientCallback = new
UndertowClientCallback(camelExchange, callback, getEndpoint(),
+ request, bodyAsByte);
+
+ // when connect succeeds or fails UndertowClientCallback will
+ // get notified on a I/O thread run by Xnio worker. The writing
+ // of request and reading of response is performed also in the
+ // callback
+ client.connect(clientCallback, uri, worker, ssl, pool, options);
+
+ // the call above will proceed on Xnio I/O thread we will
+ // notify the exchange asynchronously when the HTTP exchange
+ // ends with success or failure from UndertowClientCallback
+ return false;
}
- final UndertowClientCallback clientCallback = new
UndertowClientCallback(camelExchange, callback, getEndpoint(),
- request, bodyAsByte);
+ }
- // when connect succeeds or fails UndertowClientCallback will
- // get notified on a I/O thread run by Xnio worker. The writing
- // of request and reading of response is performed also in the
- // callback
- client.connect(clientCallback, uri, worker, ssl, pool, options);
+ private boolean processWebSocket(final Exchange camelExchange, final
AsyncCallback camelCallback) {
+ final Message in = camelExchange.getIn();
+ try {
+ Object message = in.getBody();
+ if (!(message instanceof String || message instanceof byte[] ||
message instanceof Reader
+ || message instanceof InputStream)) {
+ message = in.getBody(String.class);
+ }
- // the call above will proceed on Xnio I/O thread we will
- // notify the exchange asynchronously when the HTTP exchange
- // ends with success or failure from UndertowClientCallback
- return false;
+ if (message != null) {
+ final int timeout = endpoint.getSendTimeout();
+ if (isSendToAll(in)) {
+ return webSocketHandler.send(peer -> true, message,
timeout, camelExchange, camelCallback);
+ }
+ final List<String> connectionKeys =
in.getHeader(UndertowConstants.CONNECTION_KEY_LIST, List.class);
+ if (connectionKeys != null) {
+ return webSocketHandler.send(
+ peer ->
connectionKeys.contains(peer.getAttribute(UndertowConstants.CONNECTION_KEY)),
message,
+ timeout, camelExchange, camelCallback);
+ }
+ final String connectionKey =
in.getHeader(UndertowConstants.CONNECTION_KEY, String.class);
+ if (connectionKey != null) {
+ return webSocketHandler.send(
+ peer ->
connectionKey.equals(peer.getAttribute(UndertowConstants.CONNECTION_KEY)),
message,
+ timeout, camelExchange, camelCallback);
+ }
+ throw new IllegalStateException(
+ String.format("Cannot process message which has none
of the headers %s, %s or %s set: %s",
+ UndertowConstants.SEND_TO_ALL,
UndertowConstants.CONNECTION_KEY_LIST,
+ UndertowConstants.CONNECTION_KEY, in));
+ } else {
+ /* nothing to do for a null body */
+ camelCallback.done(true);
+ return true;
+ }
+ } catch (IOException e) {
Review comment:
Catch all kinds of exceptions, eg Exception or Throwable
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Add websocket support to camel-undertow
> ---------------------------------------
>
> Key: CAMEL-12005
> URL: https://issues.apache.org/jira/browse/CAMEL-12005
> Project: Camel
> Issue Type: New Feature
> Reporter: Thomas Diesler
> Assignee: Peter Palaga
>
> CrossRef: https://github.com/wildfly-extras/wildfly-camel/issues/2298
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
