docs: @Override
      public void onResponse(ReadResultSet<ReliableTopicMessage> result) {
          // we process all messages in batch. So we don't release the thread and reschedule ourselves;
          // but we'll process whatever was received in 1 go.
          for (Object item : result) {
              ReliableTopicMessage message = (ReliableTopicMessage) item;
  
              if (cancelled) {
                  return;
              }
  
              try {
                  listener.storeSequence(sequence);
                  process(message);
              } catch (Throwable t) {
                  if (terminate(t)) {
                      cancel();
                      return;
                  }
              }
  
              sequence++;
          }
          next();
      }