MULE- Как прослушивать клиентский сокет, созданный исходящей конечной точкой

Question

У меня есть внешний TCP-сервер, который прослушивает порт и как только соединение успешно выполняется клиентом, оно начнет толкать данные (можно считать это типичным публикация-подписка). Только для первоначального запроса из приложения на сервер для создания соединения сокета другие запросы не будут отправляться. Сервер подталкивает данные всякий раз, когда он имеет. Проблема заключается в том, что я установил соединение с сервером с использованием исходящей конечной точки TCP, но как я могу непрерывно прослушивать сокет, созданный моим исходящим, для получения данных, опубликованных сервером?

Answer 1

Один из способов разделить сокет подключения клиента к слушателю будет использовать пользовательский MessageDispatcher в соединителе исходящих TCP, как -

<tcp:connector name="TCP2" doc:name="TCP connector" 
    clientSoTimeout="70000" receiveBacklog="0" receiveBufferSize="0" 
    sendBufferSize="0" serverSoTimeout="70000" socketSoLinger="0" 
    validateConnections="true" keepAlive="true" sendTcpNoDelay="true" 
    keepSendSocketOpen="true"> 
    <receiver-threading-profile 
     maxThreadsActive="1" maxThreadsIdle="1" /> 
    <reconnect-forever /> 
    <service-overrides dispatcherFactory="CustomMessageDispatcherFactory"/> 
</tcp:connector> 

Вы будете иметь диспетчерскую класс фабрики как

import org.mule.api.MuleException; 
    import org.mule.api.endpoint.OutboundEndpoint; 
    import org.mule.api.transport.MessageDispatcher; 
    import org.mule.transport.tcp.TcpMessageDispatcherFactory; 

    public class CustomMessageDispatcherFactory extends TcpMessageDispatcherFactory { 

     public MessageDispatcher create(OutboundEndpoint endpoint) throws MuleException 
     { 
      return new CustomMessageDispatcher(endpoint); 
     } 
    } 

И CustomMessageDispatcher.class, как показано ниже -

import org.mule.api.MuleEvent; 
    import org.mule.api.MuleMessage; 
    import org.mule.api.endpoint.OutboundEndpoint; 
    import org.mule.api.transformer.TransformerException; 
    import org.mule.transport.AbstractMessageDispatcher; 
    import org.mule.transport.NullPayload; 
    import org.mule.transport.tcp.TcpConnector;   
    import java.io.BufferedOutputStream; 
    import java.io.IOException; 
    import java.net.Socket; 

    /** 
    * Send transformed Mule events over TCP. 
    */ 
    public class CustomMessageDispatcher extends AbstractMessageDispatcher 
    {  
     private final TcpConnector connector;  

     public CustomMessageDispatcher (OutboundEndpoint endpoint) 
     { 
      super(endpoint); 
      this.connector = (TcpConnector) endpoint.getConnector(); 
     } 

     @Override 
    protected synchronized void doDispatch(MuleEvent event) throws Exception 
    { 
/* Share the socket with the mule flow as a session variable */ 
     Socket socket = connector.getSocket(endpoint); 
     event.getMessage().setInvocationProperty("ClientSocket", socket); 

/* If you have something to be dispatched, you can use the below section of code */ 
     try 
     { 
      dispatchToSocket(socket, event); 
     } 
     finally 
     { 
      connector.releaseSocket(socket, endpoint); 
     } 
    } 

     @Override 
    protected MuleMessage doSend(MuleEvent event) throws Exception { 
     // Not used since we do not do request-response for the outbound endpoint 
     return null; 

    }  

    } 

I Если вы хотите слушать один и тот же сокет, вы можете использовать тот же класс для прослушивания. Вы можете написать собственную реализацию для метода doSend(). Этот метод запускается, когда конечная точка устанавливается как «запрос-ответ», и он получает массив байтов и отправляет его обратно в поток мула после назначения его в качестве полезной нагрузки. Пожалуйста, обратитесь к классу org.mule.transport.tcp.TcpMessageDispatcher.class за использование этих методов по умолчанию. FYI - Я не тестировал эту логику самостоятельно, но у меня есть аналогичная реализация для отправки объекта сокета через потоки мула. Надеюсь, это вам поможет.