@Override
  public synchronized void subscribe(final Subscriber<Collection<StreamedRow>> subscriber) {
    final PullQuerySubscription subscription = new PullQuerySubscription(
        exec, subscriber, result);

    result.start();

    final WebSocketSubscriber<StreamedRow> webSocketSubscriber =
        (WebSocketSubscriber<StreamedRow>) subscriber;

    webSocketSubscriber.onSubscribe(subscription, metricsCallbackHolder, startTimeNanos);
  }
@Test
  public void shouldSubscribe() {
    // When:
    publisher.subscribe(subscriber);

    // Then:
    verify(subscriber).onSubscribe(any(), any(), anyLong());
  }