大气WebSocketProtocol广播

Question

我有一个工作的基于jetty的WebSocket服务器，我需要迁移到使用Atmosphere。我需要将传入的WebSockets消息写入内部队列以进行后续处理，并将原始消息作为对事务的积极确认返回给客户端。

我已经实现使用的WebSocketProtocol API的气氛。连接成功启动，一条消息从客户端发送到服务器。然而，没有收到任何响应，似乎连接已关闭。

下面是连接请求的日志消息:

INFO  [2015-05-19 14:46:08,222] org.eclipse.jetty.server.ServerConnector: Started application@1db9d253{HTTP/1.1}{0.0.0.0:9090}
INFO  [2015-05-19 14:46:08,236] org.eclipse.jetty.server.ServerConnector: Started admin@6f5f892c{HTTP/1.1}{0.0.0.0:8081}
INFO  [2015-05-19 14:46:08,237] org.eclipse.jetty.server.Server: Started @15902ms
INFO  [2015-05-19 14:46:08,344] com.df.sensoringest.services.SensorIngestServiceIntegrationTest: Connecting to: ws://localhost:9090/dF/sensorIngest
127.0.0.1 - - [19/May/2015:14:46:08 +0000] "GET /dF/sensorIngest HTTP/1.1" 101 - "-" "-"  259
INFO  [2015-05-19 14:46:09,014] com.df.sensoringest.services.SensorIngestServiceIntegrationTest: Connected: WebSocketSession[websocket=JettyListenerEventDriver[com.df.sensoringest.services.SensorIngestServiceIntegrationTest$TestClient],behavior=CLIENT,connection=WebSocketClientConnection@79056e19{IDLE}{f=Flusher[queueSize=0,aggregateSize=0,failure=null],g=Generator[CLIENT,validating],p=Parser@2b100b71[ExtensionStack,s=START,c=0,len=0,f=null,p=WebSocketPolicy@6c4a33a4[behavior=CLIENT,maxTextMessageSize=65536,maxTextMessageBufferSize=32768,maxBinaryMessageSize=65536,maxBinaryMessageBufferSize=32768,asyncWriteTimeout=60000,idleTimeout=300000,inputBufferSize=4096]]},remote=WebSocketRemoteEndpoint@432bf2ff[batching=true],incoming=JettyListenerEventDriver[com.df.sensoringest.services.SensorIngestServiceIntegrationTest$TestClient],outgoing=ExtensionStack[queueSize=0,extensions=[],incoming=org.eclipse.jetty.websocket.common.WebSocketSession,outgoing=org.eclipse.jetty.websocket.client.io.WebSocketClientConnection]]
INFO  [2015-05-19 14:46:09,032] com.df.sensoringest.resources.socket.SocketResource: connection requested: 4c3b4254-a3bc-4a1f-900b-3b8ad9b9afc6

连接建立后，我发送一条消息，它被WebSocket服务器接收:

INFO  [2015-05-19 14:46:16,015] com.df.sensoringest.services.SensorIngestServiceIntegrationTest: sending: This is a test message
INFO  [2015-05-19 14:46:16,035] com.df.sensoringest.resources.socket.SocketResource: received: This is a test message
INFO  [2015-05-19 14:46:16,352] com.df.sensoringest.resources.socket.SocketResource: connection closed: 4c3b4254-a3bc-4a1f-900b-3b8ad9b9afc6

客户端从未接收到响应，连接被关闭。没有抛出异常，所以我不确定发生了什么。

以下是WebSocket服务器代码:

@Slf4j
@WebSocketProtocolService
public class SocketResource implements WebSocketProtocol {
  @Override
  public List<AtmosphereRequest> onMessage(WebSocket webSocket, String data) {
    log.info("received: {}", data);
    webSocket.resource().getBroadcaster().broadcast(data);
    return null;
  }
  @Override
  public List<AtmosphereRequest>
      onMessage(WebSocket webSocket, byte[] data, int offset, int length) {
    try {
      String bStr = new String(data, "UTF-8");
      log.info("received unexpected byte message: {}", bStr);
    } catch (UnsupportedEncodingException e) {
      log.error("unsupported text encoding: {}", e);
    }
    return null;
  }
  @Override
  public void configure(AtmosphereConfig config) {
    log.info("config method called");
  }
  @Override
  public void onOpen(WebSocket webSocket) {
    log.info("connection requested: {}", webSocket.resource().uuid());
  }
  @Override
  public void onClose(WebSocket webSocket) {
    log.info("connection closed: {}", webSocket.resource().uuid());
  }
  @Override
  public void onError(WebSocket webSocket, WebSocketProcessor.WebSocketException t) {
    log.error("websocket error: {}", t.getMessage());
    log.error("status {}, message {}", t.response().getStatus(), t.response().getStatusMessage());
  }
}

下面是测试客户端:

  /**
   * The Test WebSockets Client
   */
  public static class TestClient extends WebSocketAdapter {
    @Override
    public void onWebSocketClose(int statusCode, String reason) {
      log.error("websocket session closed: {}, {}", statusCode, reason);
    }
    @Override
    public void onWebSocketConnect(Session session) {
      log.info("Connected: {}", session);
      try {
        Thread.sleep((long)(5 * DateTimeConstants.MILLIS_PER_SECOND));
      } catch (Exception e) {
        log.error("sleep interrupted ...", e);
      }
      try {
        log.info("sending: {}", sensorMessage);
        session.getRemote().sendString(sensorMessage);
      } catch (Exception e) {
        log.error("error sending message: {}", e);
      }
    }
    @Override
    public void onWebSocketError(Throwable cause) {
      log.info("unexpected websockets exception: {}", cause);
    }
    @Override
    public void onWebSocketBinary(byte[] payload, int offset, int len) {
      log.info("unexpected binary message received");
    }
    @Override
    public void onWebSocketText(String message) {
      if (message != null) {
        response = message;
        log.info("websocket response received: {}", message);
      }
    }
  }

那么，在所有这一切之后，我做错了什么或没有做什么导致连接关闭和广播回客户端失败?

Answer 1

我没有在onOpen()方法中设置广播器。在这样做之后，并在我的onMessage()方法中获得广播器，现在我得到了我在客户端中寻找的响应。

下面是新的套接字资源代码:

/**
 * Example resource to show integration of Atmosphere within Dropwizard.
 */
@Slf4j
@WebSocketProtocolService
public class SocketResource implements WebSocketProtocol {
  @Inject
  private BroadcasterFactory factory;
  private int requestCount = 1;
  @Override
  public List<AtmosphereRequest> onMessage(WebSocket webSocket, String data) {
    Broadcaster b = webSocket.resource().getBroadcaster();
    b.broadcast(data);
    log.info("broadcaster: {}", b);
    log.info("received: {}", data);
    return null;
  }
  @Override
  public List<AtmosphereRequest>
      onMessage(WebSocket webSocket, byte[] data, int offset, int length) {
    try {
      String bStr = new String(data, "UTF-8");
      log.info("received unexpected byte message: {}", bStr);
    } catch (UnsupportedEncodingException e) {
      log.error("unsupported text encoding: {}", e);
    }
    return null;
  }
  @Override
  public void configure(AtmosphereConfig config) {
    log.info("config method called");
  }
  @Override
  public void onOpen(WebSocket webSocket) {
    log.info("connection requested: {}", webSocket.resource().uuid());
    webSocket.resource().setBroadcaster(factory.lookup("/sensorIngest/" +
        requestCount++, true)).suspend(-1);
  }
  @Override
  public void onClose(WebSocket webSocket) {
    log.info("connection closed: {}", webSocket.resource().uuid());
  }
  @Override
  public void onError(WebSocket webSocket, WebSocketProcessor.WebSocketException t) {
    log.error("websocket error: {}", t.getMessage());
    log.error("status {}, message {}", t.response().getStatus(), t.response().getStatusMessage());
  }

我是开放的改进，因为我真的是新的气氛。我还没有弄清楚的一件事是如何将服务器响应仅发送到一个客户端。