Ratpack publishing to websocket

gistfile1.java

public class MetricsWebsocketBroadcastHandler implements Handler {

  @Override
  public void handle(final Context context) throws Exception {
    final MetricsPublisher publisher = context.get(MetricsPublisher.class);

    websocket(context, new AutoCloseWebSocketHandler<AutoCloseable>() {
      @Override
      public AutoCloseable onOpen(final WebSocket webSocket) throws Exception {
        MetricsWebsocketSubscriber subscriber = new MetricsWebsocketSubscriber(webSocket);
        context.stream(publisher, subscriber);
        return subscriber;
      }
    });
  }
}


public class MetricsWebsocketSubscriber implements Subscriber<String>, AutoCloseable {
  private final WebSocket webSocket;
  private Subscription subscription;

  public MetricsWebsocketSubscriber(WebSocket webSocket) {
    this.webSocket = webSocket;
  }

  @Override
  public void close() {
    if (subscription != null) {
      this.subscription.cancel();
    }
  }

  @Override
  public void onSubscribe(Subscription s) {
    if (this.subscription == null) {
      this.subscription = s;
      this.subscription.request(Integer.MAX_VALUE);
    } else {
      this.subscription.cancel();
    }
  }

  @Override
  public void onNext(String s) {
    webSocket.send(s);
  }

  @Override
  public void onError(Throwable t) {
    webSocket.close();
  }

  @Override
  public void onComplete() {
    webSocket.close();
  }
}