title: Spring Boot SSL TCPClient ~ StompBrokerRelayMessageHandler ~ ActiveMQ ~ Undertow
body: <p>I'm attempting to build a websocket messaging app based on the <a href="http://spring.io/guides/gs/messaging-stomp-websocket/" rel="noreferrer">Spring Websocket Demo</a> running <a href="http://activemq.apache.org/" rel="noreferrer">ActiveMQ</a> as the STOMP message broker with <a href="http://undertow.io/" rel="noreferrer">Undertow</a>. The application runs fine on insecure connections.  However, I'm having difficulty configuring the <a href="http://docs.spring.io/spring/docs/current/spring-framework-reference/html/websocket.html#websocket-stomp-handle-broker-relay-configure" rel="noreferrer">STOMP Broker Relay</a> to forward with SSL connections. </p>
  
  <p>As mentioned in the Spring WebSocket Docs... </p>
  
  <blockquote>
    <p>The "STOMP broker relay" in the above configuration is a Spring MessageHandler that handles messages by forwarding them to an external message broker. To do so it establishes TCP connections to the broker, forwards all messages to it, and then forwards all messages received from the broker to clients through their WebSocket sessions. Essentially it acts as a "relay" that forwards messages in both directions.</p>
  </blockquote>
  
  <p>Further, the docs state a dependency on <a href="https://github.com/reactor/reactor" rel="noreferrer">reactor-net</a> which I have... </p>
  
  <blockquote>
    <p>Please add a dependency on org.projectreactor:reactor-net for TCP connection management.</p>
  </blockquote>
  
  <p>The issue is that my current implementation doesn't initialize the <a href="https://github.com/reactor/reactor/blob/v2.0.7.RELEASE/reactor-net/src/main/java/reactor/io/net/impl/netty/tcp/NettyTcpClient.java" rel="noreferrer">NettyTCPClient</a> via SSL so the ActiveMQ connection fails with an SSLException.</p>
  
  <hr>
  
  <pre><code>[r.i.n.i.n.t.NettyTcpClient:307] » CONNECTED: 
  [id: 0xcfef39e9, /127.0.0.1:17779 =&gt; localhost/127.0.0.1:8442]
  ...
  [o.a.a.b.TransportConnection.Transport:245] » 
  Transport Connection to: tcp://127.0.0.1:17779 failed:
  javax.net.ssl.SSLException: Unrecognized SSL message, plaintext connection?
  ...
  </code></pre>
  
  <p>As such I've attempted to research the <a href="https://github.com/reactor/reactor" rel="noreferrer">Project Reactor Docs</a> to set SSL options for the connection but I haven't been successful. </p>
  
  <p>At this point I've found the <a href="https://github.com/spring-projects/spring-framework/blob/master/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java" rel="noreferrer">StompBrokerRelayMessageHandler</a> initializes the <a href="https://github.com/reactor/reactor/blob/v2.0.7.RELEASE/reactor-net/src/main/java/reactor/io/net/impl/netty/tcp/NettyTcpClient.java" rel="noreferrer">NettyTCPClient</a> by default in <a href="https://github.com/spring-projects/spring-framework/blob/master/spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompBrokerRelayMessageHandler.java#L957" rel="noreferrer">Reactor2TcpClient</a> yet, it doesn't appear to configurable.</p>
  
  <p>Assistance would be greatly appreciated.</p>
  
  <p><em>SSCCE</em></p>
  
  <hr>
  
  <blockquote>
    <p>app.props</p>
  </blockquote>
  
  <pre><code>spring.activemq.in-memory=true
  spring.activemq.pooled=false
  spring.activemq.broker-url=stomp+ssl://localhost:8442
  server.port=8443
  server.ssl.enabled=true
  server.ssl.protocol=tls
  server.ssl.key-alias=undertow
  server.ssl.key-store=classpath:undertow.jks
  server.ssl.key-store-password=xxx
  server.ssl.trust-store=classpath:undertow_certs.jks
  server.ssl.trust-store-password=xxx
  </code></pre>
  
  <blockquote>
    <p>WebSocketConfig</p>
  </blockquote>
  
  <pre><code>//... 
  @Configuration
  @EnableWebSocketMessageBroker
  public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer {
  
      private static final Logger log = LoggerFactory.getLogger(WebSocketConfig.class);
  
      private final static String KEYSTORE = "/activemq.jks";
      private final static String KEYSTORE_PASS = "xxx";
      private final static String KEYSTORE_TYPE = "JKS";
      private final static String TRUSTSTORE = "/activemq_certs.jks";
      private final static String TRUSTSTORE_PASS = "xxx";
  
      private static String getBindLocation() {
          return "stomp+ssl://localhost:8442?transport.needClientAuth=false";
      }
  
      @Bean(initMethod = "start", destroyMethod = "stop")
      public SslBrokerService activeMQBroker() throws Exception {
  
          final SslBrokerService service = new SslBrokerService();
          service.setPersistent(false);
  
          KeyManager[] km = SecurityManager.getKeyManager();
          TrustManager[] tm = SecurityManager.getTrustManager();
  
          service.addSslConnector(getBindLocation(), km, tm, null);
          final ActiveMQTopic topic = new ActiveMQTopic("jms.topic.test");
          service.setDestinations(new ActiveMQDestination[]{topic});
  
          return service;
      }
  
  
      @Override
      public void configureMessageBroker(MessageBrokerRegistry config) {
          config.enableStompBrokerRelay("/topic").setRelayHost("localhost").setRelayPort(8442);
          config.setApplicationDestinationPrefixes("/app");
      }
  
      @Override
      public void registerStompEndpoints(StompEndpointRegistry registry) {
          registry.addEndpoint("/welcome").withSockJS();
          registry.addEndpoint("/test").withSockJS();
      }
  
     private static class SecurityManager { 
     //elided...
     }
  
  }
  </code></pre>
  
  <blockquote>
    <p>SOLVED Per <a href="https://stackoverflow.com/users/480980/rossen-stoyanchev">Rossens</a> Advice. Here's the implementation details for anyone interested.</p>
  </blockquote>
  
  <hr>
  
  <blockquote>
    <p>WebSocketConfig </p>
  </blockquote>
  
  <pre><code>@Configuration
  public class WebSocketConfig extends DelegatingWebSocketMessageBrokerConfiguration {
      ...
      @Bean
      public AbstractBrokerMessageHandler stompBrokerRelayMessageHandler() {
        StompBrokerRelayMessageHandler handler = (StompBrokerRelayMessageHandler) super.stompBrokerRelayMessageHandler();
        ConfigurationReader reader = new StompClientDispatcherConfigReader();
        Environment environment = new Environment(reader).assignErrorJournal();
        TcpOperations&lt;byte[]&gt; client = new Reactor2TcpClient&lt;&gt;(new StompTcpClientSpecFactory(environment,"localhost", 8443));
        handler.setTcpClient(client);
        return handler;
      }
  }
  </code></pre>
  
  <blockquote>
    <p>StompTCPClientSpecFactory</p>
  </blockquote>
  
  <pre><code>private static class StompTcpClientSpecFactory
          implements NetStreams.TcpClientFactory&lt;Message&lt;byte[]&gt;, Message&lt;byte[]&gt;&gt; {
  
      private static final Logger log = LoggerFactory.getLogger(StompTcpClientSpecFactory.class);
  
      private final String host;
      private final int port;
      private final String KEYSTORE = "src/main/resources/tcpclient.jks";
      private final String KEYSTORE_PASS = "xxx";
      private final String KEYSTORE_TYPE = "JKS";
      private final String TRUSTSTORE = "/src/main/resources/tcpclient_certs.jks";
      private final String TRUSTSTORE_PASS = "xxx";
      private final String TRUSTSTORE_TYPE = "JKS";
      private final Environment environment;
  
      private final SecurityManager tcpManager = new SecurityManager
              .SSLBuilder(KEYSTORE, KEYSTORE_PASS)
              .keyStoreType(KEYSTORE_TYPE)
              .trustStore(TRUSTSTORE, TRUSTSTORE_PASS)
              .trustStoreType(TRUSTSTORE_TYPE)
              .build();
  
      public StompTcpClientSpecFactory(Environment environment, String host, int port) {
          this.environment = environment;
          this.host = host;
          this.port = port;
      }
  
      @Override
      public Spec.TcpClientSpec&lt;Message&lt;byte[]&gt;, Message&lt;byte[]&gt;&gt; apply(
              Spec.TcpClientSpec&lt;Message&lt;byte[]&gt;, Message&lt;byte[]&gt;&gt; tcpClientSpec) {
  
          return tcpClientSpec
                  .ssl(new SslOptions()
                          .sslProtocol("TLS")
                          .keystoreFile(tcpManager.getKeyStore())
                          .keystorePasswd(tcpManager.getKeyStorePass())
                          .trustManagers(tcpManager::getTrustManager)
                          .trustManagerPasswd(tcpManager.getTrustStorePass()))
                  .codec(new Reactor2StompCodec(new StompEncoder(), new StompDecoder()))
                  .env(this.environment)
                  .dispatcher(this.environment.getCachedDispatchers("StompClient").get())
                  .connect(this.host, this.port);
      }
  }
  </code></pre>