OkHttp Server-Sent Events (SSE)

import okhttp3.sse.EventSource
import okhttp3.sse.EventSourceListener
import okhttp3.sse.EventSources
import okhttp3.OkHttpClient
import okhttp3.Request

import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;
import okhttp3.OkHttpClient;
import okhttp3.Request;

import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.Response
import okhttp3.sse.EventSource
import okhttp3.sse.EventSourceListener
import okhttp3.sse.EventSources

// Create HTTP client and event source factory
val client = OkHttpClient()
val factory = EventSources.createFactory(client)

// Create event listener
val listener = object : EventSourceListener() {
    override fun onOpen(eventSource: EventSource, response: Response) {
        println("Connection opened")
    }
    
    override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {
        println("Received event: $data")
    }
    
    override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {
        println("Connection failed: ${t?.message}")
    }
    
    override fun onClosed(eventSource: EventSource) {
        println("Connection closed")
    }
}

// Create and start event source
val request = Request.Builder()
    .url("https://example.com/events")
    .build()
    
val eventSource = factory.newEventSource(request, listener)

// Clean up resources when done
eventSource.cancel()

/**
 * Creates a factory for event sources using the provided OkHttpClient.
 * Automatically adds "Accept: text/event-stream" header if not present.
 */
@JvmStatic
fun createFactory(client: OkHttpClient): EventSource.Factory

/**
 * Processes an existing Response as server-sent events.
 * Useful when you already have a Response object from an HTTP call.
 */
@JvmStatic
fun processResponse(response: Response, listener: EventSourceListener)

interface EventSource {
    /** Returns the original request that initiated this event source */
    fun request(): Request
    
    /** Immediately and violently release resources held by this event source */
    fun cancel()
    
    /** Factory interface for creating new event sources */
    fun interface Factory {
        /** Creates a new event source and immediately returns it */
        fun newEventSource(request: Request, listener: EventSourceListener): EventSource
    }
}

abstract class EventSourceListener {
    /**
     * Invoked when an event source has been accepted by the remote peer 
     * and may begin transmitting events.
     */
    open fun onOpen(eventSource: EventSource, response: Response) {}
    
    /**
     * Invoked when a server-sent event is received.
     * 
     * Server-sent events have the format:
     * - id: optional unique identifier for the event
     * - event: optional event type name 
     * - data: the actual event payload/message
     * 
     * @param eventSource The event source that received the event
     * @param id Optional event ID from the 'id:' field (null if not specified)
     * @param type Optional event type from the 'event:' field (null if not specified)
     * @param data The event data from the 'data:' field
     */
    open fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {}
    
    /**
     * Invoked when the event source has been closed normally by the server.
     * 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.
     * Incoming events may have been lost. No further calls to this listener will be made.
     * 
     * @param eventSource The event source that failed
     * @param t The throwable that caused the failure (null for certain types of failures)
     * @param response The HTTP response if available (null for network failures)
     */
    open fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {}
}

import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;

// Create client and factory
OkHttpClient client = new OkHttpClient();
EventSource.Factory factory = EventSources.createFactory(client);

// Create listener
EventSourceListener listener = new EventSourceListener() {
    @Override
    public void onOpen(EventSource eventSource, Response response) {
        System.out.println("Connection opened");
    }
    
    @Override
    public void onEvent(EventSource eventSource, String id, String type, String data) {
        System.out.println("Event: " + data);
    }
    
    @Override
    public void onFailure(EventSource eventSource, Throwable t, Response response) {
        System.out.println("Error: " + (t != null ? t.getMessage() : "Unknown"));
    }
    
    @Override
    public void onClosed(EventSource eventSource) {
        System.out.println("Connection closed");
    }
};

// Create event source
Request request = new Request.Builder()
    .url("https://example.com/events")
    .build();
    
EventSource eventSource = factory.newEventSource(request, listener);

// Clean up
eventSource.cancel();

import okhttp3.sse.EventSources

// When you already have a Response object from an HTTP call
val response = client.newCall(request).execute()

// Process the response as SSE
EventSources.processResponse(response, listener)

import java.util.concurrent.TimeUnit

// Add custom headers to the request
val request = Request.Builder()
    .url("https://example.com/events")
    .header("Authorization", "Bearer token")
    .header("Custom-Header", "value")
    .build()

// Configure OkHttp client with custom settings
val client = OkHttpClient.Builder()
    .connectTimeout(30, TimeUnit.SECONDS)
    .readTimeout(60, TimeUnit.SECONDS)
    .build()

val factory = EventSources.createFactory(client)
val eventSource = factory.newEventSource(request, listener)

override fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?) {
    when {
        t is IOException && response == null -> {
            println("Network connection error: ${t.message}")
        }
        response != null -> {
            when (response.code) {
                401 -> println("Authentication failed")
                404 -> println("SSE endpoint not found")
                in 400..499 -> println("Client error: ${response.code} ${response.message}")
                in 500..599 -> println("Server error: ${response.code} ${response.message}")
                else -> println("HTTP error: ${response.code} ${response.message}")
            }
        }
        t != null -> println("Parsing or protocol error: ${t.message}")
        else -> println("Unknown error occurred")
    }
}

// Core SSE API types
interface EventSource {
    fun request(): Request
    fun cancel()
    
    fun interface Factory {
        fun newEventSource(request: Request, listener: EventSourceListener): EventSource
    }
}

abstract class EventSourceListener {
    open fun onOpen(eventSource: EventSource, response: Response)
    open fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String)
    open fun onClosed(eventSource: EventSource)
    open fun onFailure(eventSource: EventSource, t: Throwable?, response: Response?)
}

object EventSources {
    @JvmStatic fun createFactory(client: OkHttpClient): EventSource.Factory
    @JvmStatic fun processResponse(response: Response, listener: EventSourceListener)
}

// Required OkHttp types (from okhttp3 package)
class OkHttpClient
class Request  
class Response