data: String,
  ) {
    get().log("[ES] onEvent", Platform.INFO, null)
    events.add(Event(id, type, data))
    drainCancelQueue(eventSource)
  }

  override fun onClosed(eventSource: EventSource) {
    get().log("[ES] onClosed", Platform.INFO, null)
    events.add(Closed)
    drainCancelQueue(eventSource)
  }

  override fun onFailure(
    eventSource: EventSource,
    t: Throwable?,

 * limitations under the License.
 */
package okhttp3

import okio.IOException

/**
 * Listener for connection events. Extend this class to monitor the new connections and closes.
 *
 * All event methods must execute fast, without external locking, cannot throw exceptions,
 * attempt to mutate the event parameters, or be reentrant back into the client.

  }

  /**
   * Invoked when both peers have indicated that no more messages will be transmitted and the
   * connection has been successfully released. No further calls to this listener will be made.
   */
  open fun onClosed(
    webSocket: WebSocket,
    code: Int,
    reason: String,
  ) {
  }

  /**
   * Invoked when a web socket has been closed due to an error reading from or writing to the

    if (eventSequence.isEmpty()) {
      assertThat(e).matchesPredicate { it is CallStart || it is Canceled }
    } else {
      eventSequence.forEach loop@{
        when (e.closes(it)) {
          null -> return // no open event
          true -> return // found open event
          false -> return@loop // this is not the open event so continue
        }
      }

      events.add(Closing(code, reason))
    }
  }

  override fun onClosed(
    webSocket: WebSocket,
    code: Int,
    reason: String,
  ) {
    Platform.get().log("[WS $name] onClosed $code", Platform.INFO, null)
    val delegate = delegate
    if (delegate != null) {
      this.delegate = null
      delegate.onClosed(webSocket, code, reason)
    } else {
      events.add(Closed(code, reason))
    }

@SuppressSignatureCheck
sealed class ConnectionEvent {
  abstract val timestampNs: Long
  open val connection: Connection?
    get() = null

  /** Returns if the event closes this event, or null if this is no open event. */
  open fun closes(event: ConnectionEvent): Boolean? = null

  val name: String
    get() = javaClass.simpleName

  data class ConnectStart(
    override val timestampNs: Long,
    val route: Route,

	public abstract fun newEventSource (Lokhttp3/Request;Lokhttp3/sse/EventSourceListener;)Lokhttp3/sse/EventSource;
}

public abstract class okhttp3/sse/EventSourceListener {
	public fun <init> ()V
	public fun onClosed (Lokhttp3/sse/EventSource;)V
	public fun onEvent (Lokhttp3/sse/EventSource;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)V
	public fun onFailure (Lokhttp3/sse/EventSource;Ljava/lang/Throwable;Lokhttp3/Response;)V

    id: String?,
    type: String?,
    data: String,
  ) {
  }

  /**
   * TODO description.
   *
   * No further calls to this listener will be made.
   */
  open fun onClosed(eventSource: EventSource) {
  }

  /**
   * Invoked when an event source has been closed due to an error reading from or writing to the

          }
        listener.onFailure(this, exception, response)
        return
      }
      if (canceled) {
        listener.onFailure(this, IOException("canceled"), response)
      } else {
        listener.onClosed(this)
      }
    }
  }

  private fun ResponseBody.isEventStream(): Boolean {
    val contentType = contentType() ?: return false
    return contentType.type == "text" && contentType.subtype == "event-stream"

    assertThat(relay.isClosed).isTrue()
    assertFile(Relay.PREFIX_CLEAN, 20L, metadata.size, "abcdefghijklmnopqrst", metadata)
  }

  /** Returns a callable that reads all of source, closes it, and returns the bytes.  */
  private fun sourceReader(source: Source?): Callable<ByteString> {
    return Callable {
      val buffer = Buffer()
      while (source!!.read(buffer, 16384) != -1L) {
      }