val event = nextEvent() as Open
    return event.eventSource
  }

  fun assertClose() {
    nextEvent() as Closed
  }

  fun assertFailure(message: String?) {
    val event = nextEvent() as Failure
    if (message != null) {
      assertThat(event.t!!.message).isEqualTo(message)
    } else {
      assertThat(event.t).isNull()
    }
  }

  internal data class Open(

    assertThat(nextEvent()).isEqualTo(Ping(payload))
  }

  fun assertPong(payload: ByteString) {
    assertThat(nextEvent()).isEqualTo(Pong(payload))
  }

  fun assertClosing(
    code: Int,
    reason: String,
  ) {
    assertThat(nextEvent()).isEqualTo(Closing(code, reason))
  }

  fun assertClosed(
    code: Int,
    reason: String,
  ) {

		qs.events[last].Data[0] = 0
	}
	if qs.nextEvent >= len(qs.events) && qs.waitingForDrain {
		qs.waitingForDrain = false
		<-qs.signalc
		<-qs.blockedc
	}
	if qs.nextEvent >= len(qs.events) {
		qs.events = qs.events[:0]
		qs.nextEvent = 0
		return QUICEvent{Kind: QUICNoEvent}
	}
	e := qs.events[qs.nextEvent]
	qs.events[qs.nextEvent] = QUICEvent{} // zero out references to data
	qs.nextEvent++

      if (!dispatching.get()) {
        dispatching.set(true);
        try {
          Event nextEvent;
          while ((nextEvent = queueForThread.poll()) != null) {
            while (nextEvent.subscribers.hasNext()) {
              nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
            }
          }
        } finally {
          dispatching.remove();
          queue.remove();

      if (!dispatching.get()) {
        dispatching.set(true);
        try {
          Event nextEvent;
          while ((nextEvent = queueForThread.poll()) != null) {
            while (nextEvent.subscribers.hasNext()) {
              nextEvent.subscribers.next().dispatchEvent(nextEvent.event);
            }
          }
        } finally {
          dispatching.remove();
          queue.remove();

            LOGGER.debug("Started listening to file system change events");
            try {
                while (consumeEvents) {
                    FileWatchEvent nextEvent = fileEvents.take();
                    if (!stopping) {
                        nextEvent.handleEvent(new FileWatchEvent.Handler() {
                            @Override

		}
		r.spillErr = err

		// Reset CPU samples cursor.
		r.cpuSamples = r.gen.cpuSamples

		// Reset frontier.
		for m, batches := range r.gen.batches {
			bc := &batchCursor{m: m}
			ok, err := bc.nextEvent(batches, r.gen.freq)
			if err != nil {
				return Event{}, err
			}
			if !ok {
				// Turns out there aren't actually any events in these batches.
				continue
			}

	for _, c := range []*testQUICConn{a, b} {
		if !c.conn.conn.quic.started {
			if err := c.conn.Start(ctx); err != nil {
				return err
			}
		}
	}
	idleCount := 0
	for {
		e := a.conn.NextEvent()
		if onEvent != nil && onEvent(e, a, b) {
			continue
		}
		switch e.Kind {
		case QUICNoEvent:
			idleCount++
			if idleCount == 2 {
				if !a.complete || !b.complete {

type batchCursor struct {
	m       ThreadID
	lastTs  Time
	idx     int       // next index into []batch
	dataOff int       // next index into batch.data
	ev      baseEvent // last read event
}

func (b *batchCursor) nextEvent(batches []batch, freq frequency) (ok bool, err error) {
	// Batches should generally always have at least one event,
	// but let's be defensive about that and accept empty batches.

pkg crypto/tls, method (*QUICConn) ConnectionState() ConnectionState #44886
pkg crypto/tls, method (*QUICConn) HandleData(QUICEncryptionLevel, []uint8) error #44886
pkg crypto/tls, method (*QUICConn) NextEvent() QUICEvent #44886
pkg crypto/tls, method (*QUICConn) SendSessionTicket(QUICSessionTicketOptions) error #60107
pkg crypto/tls, type QUICSessionTicketOptions struct #60107