Client examples

Choose a server-side Socket.IO client that supports Engine.IO v4, WebSocket-only transport, a custom transport path, and an Authorization handshake header. A raw WebSocket client or a Socket.IO server library is not a substitute. Set PHOTON_PROJECT_ID and PHOTON_API_KEY in the process environment. The key needs project:read access to that project. Keep it on the server. PHOTON_API_ORIGIN is optional and defaults to https://api.photon.codes. Every example subscribes again on each connection with the last processed startSeq. A value of 0 starts with new events; a positive value resumes after that cursor. Do not add one. See Cursors and replay.

Node.js

Install socket.io-client (tested with 4.8.3):

npm install [email protected]

This example keeps a cursor in memory. For restart recovery, load it from durable storage before connecting and persist it after each successful handler.

import { io } from "socket.io-client";

let cursor = 0;
let epoch = 0;
let pending = 0;
let queue = Promise.resolve();
const socket = io(process.env.PHOTON_API_ORIGIN ?? "https://api.photon.codes", {
  path: `/v1/projects/${process.env.PHOTON_PROJECT_ID}/events/ws`,
  addTrailingSlash: false,
  transports: ["websocket"],
  extraHeaders: { Authorization: `Bearer ${process.env.PHOTON_API_KEY}` },
  autoConnect: false,
});

socket.on("connect", () => {
  socket.timeout(10_000).emit("subscribe", { startSeq: cursor }, (error, result) => {
    if (error) socket.io.engine.close();
    else if (!result.ok) console.error(result.error.code);
  });
});
socket.on("disconnect", () => { epoch++; });
socket.on("delivery", (event, { seq }) => {
  const generation = epoch;
  if (++pending > 128) {
    pending--;
    epoch++;
    socket.io.engine.close();
    return;
  }
  queue = queue.then(async () => {
    if (generation !== epoch || seq <= cursor) return;
    await handleEvent(event); // Your idempotent application handler.
    // Persist seq here before advancing the in-memory checkpoint.
    cursor = seq;
  }).catch(error => {
    console.error("Event processing failed", error);
    if (generation === epoch) {
      epoch++;
      socket.io.engine.close();
    }
  }).finally(() => { pending--; });
});
socket.on("delivery_error", error => console.error(error.code));
socket.on("connect_error", error => console.error(error.message));
socket.connect();

The queue bounds pending handlers. If it fills, the connection restarts and unprocessed deliveries replay from the saved cursor. Socket.IO owns reconnection; do not add a second retry loop. This example does not send synchronous replies; see Reply to a webhook event.

Checkpointed examples

The examples below keep up to 128 deliveries in a queue and process one at a time. They retain queued work across reconnects and skip replayed sequences before queueing them again, including the event being processed. The in-memory last_queued / lastQueued value tracks accepted deliveries; only the processed cursor is saved or sent in a subscription. Replace handle_event, handleEvent, or HandleEvent with your idempotent application handler. They save photon-cursor only after that handler succeeds, and read it before connecting. Do not launch concurrent work from the handler and return before it finishes. Use a separate working directory and cursor file for each project and consumer, with only one process writing that file. The temporary-file rename avoids a partially written checkpoint on process restart. It is a local demo store, not a transaction with your application database or a guarantee against disk failure. For production, persist the cursor with your application state. Until the first event succeeds, there is no replay checkpoint. Restarting with 0 starts live again; it does not recover events missed before that first checkpoint. A handler or checkpoint failure stops the process before it processes later deliveries. A full queue also exits rather than dropping an event or blocking the client’s heartbeat. Diagnose the failure, then restart the process from the saved cursor. Let the library handle network reconnection; do not add a parallel retry loop. These examples do not send synchronous replies; see Reply to a webhook event.

The Java and .NET client versions below can stall after a server-sent WebSocket close frame. They need Photon’s server-side reconnect fix for prompt recovery during a rollout. Until that fix is deployed in your environment, restart a stalled process with its saved cursor. Test rollout recovery before relying on either client in production.

Python

Use python-socketio 5.18.1 with its asyncio client. The separate worker serializes deliveries; an async event callback alone does not serialize application work.

python -m pip install "python-socketio[asyncio_client]==5.18.1"

Save the following as main.py:

main.py

import asyncio
import os
from pathlib import Path

import socketio

CURSOR_FILE = Path("photon-cursor")
cursor = int(CURSOR_FILE.read_text()) if CURSOR_FILE.exists() else 0
last_queued = cursor
deliveries = asyncio.Queue(maxsize=128)
sio = socketio.AsyncClient()


async def handle_event(event):
    print(event["type"], event["id"])  # Your idempotent application handler.


@sio.event
async def connect():
    try:
        result = await sio.call("subscribe", {"startSeq": cursor}, timeout=10)
    except socketio.exceptions.TimeoutError:
        print("Subscribe timed out; restart from the saved cursor", flush=True)
        os._exit(1)
    if not result["ok"]:
        print("Subscribe failed:", result["error"]["code"], flush=True)
        os._exit(1)


@sio.on("delivery")
def delivery(event, metadata):
    global last_queued
    seq = metadata["seq"]
    if seq <= last_queued:
        return
    try:
        deliveries.put_nowait((event, seq))
        last_queued = seq
    except asyncio.QueueFull:
        print("Delivery queue full; restart from the saved cursor", flush=True)
        os._exit(1)


@sio.on("delivery_error")
def delivery_error(error):
    print("Delivery error:", error["code"])


@sio.event
def connect_error(data):
    print("Connect error:", data)


async def main():
    global cursor
    await sio.connect(
        os.environ.get("PHOTON_API_ORIGIN", "https://api.photon.codes"),
        socketio_path=f"/v1/projects/{os.environ['PHOTON_PROJECT_ID']}/events/ws",
        transports=["websocket"],
        headers={"Authorization": f"Bearer {os.environ['PHOTON_API_KEY']}"},
    )
    while True:
        event, seq = await deliveries.get()
        if seq <= cursor:
            continue
        await handle_event(event)
        CURSOR_FILE.with_suffix(".tmp").write_text(str(seq))
        CURSOR_FILE.with_suffix(".tmp").replace(CURSOR_FILE)
        cursor = seq


asyncio.run(main())

Run it from the directory where its cursor should be stored:

python main.py

Go

Use the client package zishang520/socket.io/clients/socket/v3, tested at v3.0.6. The server package alone does not connect to Photon. This client exposes path, WebSocket transport selection, and handshake headers.

go mod init photon-events
go get github.com/zishang520/socket.io/clients/socket/[email protected]
go get github.com/zishang520/socket.io/clients/engine/[email protected]
go get github.com/zishang520/socket.io/[email protected]

Save the following as main.go:

main.go

package main

import (
	"log"
	"net/http"
	"os"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/zishang520/socket.io/clients/engine/v3/transports"
	"github.com/zishang520/socket.io/clients/socket/v3"
	"github.com/zishang520/socket.io/v3/pkg/types"
)

const cursorFile = "photon-cursor"

type delivery struct {
	event map[string]any
	seq   int64
}

func handleEvent(event map[string]any) error {
	log.Println(event["type"], event["id"]) // Your idempotent application handler.
	return nil
}

func main() {
	var cursor atomic.Int64
	if saved, err := os.ReadFile(cursorFile); err == nil {
		seq, err := strconv.ParseInt(string(saved), 10, 64)
		if err != nil {
			log.Fatal(err)
		}
		cursor.Store(seq)
	} else if !os.IsNotExist(err) {
		log.Fatal(err)
	}
	origin := os.Getenv("PHOTON_API_ORIGIN")
	if origin == "" {
		origin = "https://api.photon.codes"
	}

	opts := socket.DefaultOptions()
	opts.SetPath("/v1/projects/" + os.Getenv("PHOTON_PROJECT_ID") + "/events/ws")
	opts.SetAddTrailingSlash(false)
	opts.SetTransports(types.NewSet(transports.WebSocket))
	opts.SetExtraHeaders(http.Header{"Authorization": {"Bearer " + os.Getenv("PHOTON_API_KEY")}})
	opts.SetAutoConnect(false)
	client, err := socket.Io(origin, opts)
	if err != nil {
		log.Fatal(err)
	}

	deliveries := make(chan delivery, 128)
	lastQueued := cursor.Load()
	client.On("connect", func(...any) {
		subscribe := map[string]any{"startSeq": cursor.Load()}
		client.Timeout(10*time.Second).EmitWithAck("subscribe", subscribe)(func(result []any, err error) {
			if err != nil {
				log.Fatal("subscribe timed out; restart from the saved cursor")
			} else if ack, _ := result[0].(map[string]any); ack["ok"] != true {
				log.Fatal("subscribe failed:", ack["error"])
			}
		})
	})
	client.On("delivery", func(args ...any) {
		event := args[0].(map[string]any)
		metadata := args[1].(map[string]any)
		seq := int64(metadata["seq"].(float64))
		if seq <= lastQueued {
			return
		}
		select {
		case deliveries <- delivery{event, seq}:
			lastQueued = seq
		default:
			log.Fatal("delivery queue full; restart from the saved cursor")
		}
	})
	client.On("delivery_error", func(args ...any) { log.Println("delivery error:", args[0]) })
	client.On("connect_error", func(args ...any) { log.Println("connect error:", args[0]) })
	client.Connect()

	for d := range deliveries {
		if d.seq <= cursor.Load() {
			continue
		}
		if err := handleEvent(d.event); err != nil {
			log.Fatal(err)
		}
		if err := os.WriteFile(cursorFile+".tmp", []byte(strconv.FormatInt(d.seq, 10)), 0o600); err != nil {
			log.Fatal(err)
		}
		if err := os.Rename(cursorFile+".tmp", cursorFile); err != nil {
			log.Fatal(err)
		}
		cursor.Store(d.seq)
	}
}

Run it from the directory where its cursor should be stored:

go run .

Rust

Use rust_socketio 0.6.0 with the async feature, not the Socket.IO server library socketioxide. Unlike the other clients here, this builder takes the transport path in its URL. Leave the namespace at its default /. In this version, acknowledgement arguments arrive as an array inside Payload::Text, so the subscribe result is values[0][0]. Delivery arguments arrive directly as values[0] (event) and values[1] (metadata). Create a binary crate with cargo new photon-events, then set these dependencies in Cargo.toml:

[dependencies]
futures-util = "0.3"
rust_socketio = { version = "=0.6.0", features = ["async"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }

Save the following as src/main.rs:

src/main.rs

use futures_util::FutureExt;
use rust_socketio::{
    Event, Payload, TransportType,
    asynchronous::{Client, ClientBuilder},
};
use serde_json::{Value, json};
use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};
use std::time::Duration;
use tokio::sync::mpsc;

const CURSOR_FILE: &str = "photon-cursor";

async fn handle_event(event: &Value) -> Result<(), Box<dyn std::error::Error>> {
    println!("{} {}", event["type"], event["id"]); // Your idempotent application handler.
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let saved = match std::fs::read_to_string(CURSOR_FILE) {
        Ok(text) => text.trim().parse()?,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => 0,
        Err(error) => return Err(error.into()),
    };
    let cursor = Arc::new(AtomicU64::new(saved));
    let origin = std::env::var("PHOTON_API_ORIGIN").unwrap_or("https://api.photon.codes".into());
    let project = std::env::var("PHOTON_PROJECT_ID")?;
    let key = std::env::var("PHOTON_API_KEY")?;
    let (deliveries, mut queue) = mpsc::channel::<(Value, u64)>(128);
    let mut last_queued = saved;

    let subscribe_cursor = cursor.clone();
    let _client = ClientBuilder::new(format!("{origin}/v1/projects/{project}/events/ws"))
        .transport_type(TransportType::Websocket)
        .opening_header("Authorization", format!("Bearer {key}"))
        .on(Event::Connect, move |_: Payload, socket: Client| {
            let subscribe = json!({ "startSeq": subscribe_cursor.load(Ordering::SeqCst) });
            async move {
                let on_ack = |result: Payload, _: Client| {
                    async move {
                        if let Payload::Text(values) = result {
                            let ack = &values[0][0];
                            if ack["ok"] != true {
                                eprintln!("subscribe failed: {}", ack["error"]);
                                std::process::exit(1);
                            }
                        }
                    }
                    .boxed()
                };
                let timeout = Duration::from_secs(10);
                if let Err(error) = socket
                    .emit_with_ack("subscribe", subscribe, timeout, on_ack)
                    .await
                {
                    eprintln!("subscribe failed: {error}");
                    std::process::exit(1);
                }
            }
            .boxed()
        })
        .on("delivery", move |payload: Payload, _: Client| {
            if let Payload::Text(values) = payload {
                if let Some(seq) = values[1]["seq"].as_u64() {
                    if seq > last_queued {
                        if deliveries.try_send((values[0].clone(), seq)).is_err() {
                            eprintln!("delivery queue full; restart from the saved cursor");
                            std::process::exit(1);
                        }
                        last_queued = seq;
                    }
                }
            }
            async {}.boxed()
        })
        .on("delivery_error", |error: Payload, _: Client| {
            async move { eprintln!("delivery error: {error:?}") }.boxed()
        })
        .on(Event::Error, |error: Payload, _: Client| {
            async move { eprintln!("connection error: {error:?}") }.boxed()
        })
        .connect()
        .await?;

    while let Some((event, seq)) = queue.recv().await {
        if seq <= cursor.load(Ordering::SeqCst) {
            continue;
        }
        handle_event(&event).await?;
        std::fs::write(format!("{CURSOR_FILE}.tmp"), seq.to_string())?;
        std::fs::rename(format!("{CURSOR_FILE}.tmp"), CURSOR_FILE)?;
        cursor.store(seq, Ordering::SeqCst);
    }
    Ok(())
}

Run it from the directory where its cursor should be stored:

cargo run --release

Java

Use socket.io-client-java 2.1.2 and Java 11 or later. Add this dependency to your Maven project:

<dependency>
  <groupId>io.socket</groupId>
  <artifactId>socket.io-client</artifactId>
  <version>2.1.2</version>
</dependency>

Save the following as PhotonEvents.java in your Java source directory:

PhotonEvents.java

import io.socket.client.AckWithTimeout;
import io.socket.client.IO;
import io.socket.client.Socket;
import io.socket.engineio.client.transports.WebSocket;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import org.json.JSONObject;

public class PhotonEvents {
  static final Path CURSOR_FILE = Path.of("photon-cursor");

  static void handleEvent(JSONObject event) throws Exception {
    System.out.println(event.getString("type") + " " + event.getString("id")); // Your idempotent application handler.
  }

  public static void main(String[] args) throws Exception {
    AtomicLong cursor = new AtomicLong(
        Files.exists(CURSOR_FILE) ? Long.parseLong(Files.readString(CURSOR_FILE).trim()) : 0);
    BlockingQueue<Object[]> deliveries = new ArrayBlockingQueue<>(128);
    AtomicLong lastQueued = new AtomicLong(cursor.get());
    IO.Options options = IO.Options.builder()
        .setPath("/v1/projects/" + System.getenv("PHOTON_PROJECT_ID") + "/events/ws")
        .setTransports(new String[] {WebSocket.NAME})
        .setExtraHeaders(Map.of("Authorization", List.of("Bearer " + System.getenv("PHOTON_API_KEY"))))
        .build();
    Socket socket = IO.socket(
        System.getenv().getOrDefault("PHOTON_API_ORIGIN", "https://api.photon.codes"), options);

    socket.on(Socket.EVENT_CONNECT, connected -> {
      JSONObject subscribe = new JSONObject(Map.of("startSeq", cursor.get()));
      socket.emit("subscribe", new Object[] {subscribe}, new AckWithTimeout(10_000) {
        @Override
        public void onSuccess(Object... result) {
          JSONObject ack = (JSONObject) result[0];
          if (!ack.optBoolean("ok")) {
            System.err.println("Subscribe failed: " + ack.opt("error"));
            System.exit(1);
          }
        }

        @Override
        public void onTimeout() {
          System.err.println("Subscribe timed out");
          System.exit(1);
        }
      });
    });
    socket.on("delivery", delivery -> {
      long seq = ((JSONObject) delivery[1]).optLong("seq");
      if (seq <= lastQueued.get()) return;
      if (!deliveries.offer(delivery)) {
        System.err.println("Delivery queue full; restart from the saved cursor");
        System.exit(1);
      }
      lastQueued.set(seq);
    });
    socket.on("delivery_error", error -> System.err.println("Delivery error: " + error[0]));
    socket.on(Socket.EVENT_CONNECT_ERROR, error -> System.err.println("Connect error: " + error[0]));
    socket.connect();

    try {
      while (true) {
        Object[] delivery = deliveries.take();
        long seq = ((JSONObject) delivery[1]).getLong("seq");
        if (seq <= cursor.get()) continue;
        handleEvent((JSONObject) delivery[0]);
        Path saving = Path.of("photon-cursor.tmp");
        Files.writeString(saving, Long.toString(seq));
        Files.move(saving, CURSOR_FILE, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
        cursor.set(seq);
      }
    } catch (Exception e) {
      e.printStackTrace();
      System.exit(1); // The socket's threads would otherwise keep the process alive.
    }
  }
}

Run PhotonEvents.main with your project’s Maven or IDE launcher from the directory where its cursor should be stored.

.NET

Use SocketIOClient 4.0.5. This example uses its version 4 callback API and was tested with .NET 10.

dotnet new console -n PhotonEvents
cd PhotonEvents
dotnet add package SocketIOClient --version 4.0.5

Save the following as Program.cs:

Program.cs

using System.Text.Json;
using System.Threading.Channels;
using SocketIOClient;

const string CursorFile = "photon-cursor";
long cursor = File.Exists(CursorFile) ? long.Parse(File.ReadAllText(CursorFile)) : 0;
long lastQueued = cursor;
var deliveries = Channel.CreateBounded<(JsonElement Event, long Seq)>(128);
var origin = Environment.GetEnvironmentVariable("PHOTON_API_ORIGIN") ?? "https://api.photon.codes";

var client = new SocketIO(new Uri(origin), new SocketIOOptions
{
    Path = $"/v1/projects/{Environment.GetEnvironmentVariable("PHOTON_PROJECT_ID")}/events/ws",
    Transport = SocketIOClient.Common.TransportProtocol.WebSocket,
    ReconnectionAttempts = int.MaxValue,
    ExtraHeaders = new Dictionary<string, string>
    {
        ["Authorization"] = $"Bearer {Environment.GetEnvironmentVariable("PHOTON_API_KEY")}",
    },
});

client.OnConnected += async (_, _) =>
{
    var subscribe = new { startSeq = Interlocked.Read(ref cursor) };
    await client.EmitAsync("subscribe", new object[] { subscribe }, ack =>
    {
        var result = ack.GetValue<JsonElement>(0);
        if (!result.GetProperty("ok").GetBoolean())
        {
            Console.Error.WriteLine($"Subscribe failed: {result.GetProperty("error")}");
            Environment.Exit(1);
        }
        return Task.CompletedTask;
    });
};
client.On("delivery", context =>
{
    var seq = context.GetValue<JsonElement>(1).GetProperty("seq").GetInt64();
    if (seq <= lastQueued) return Task.CompletedTask;
    if (!deliveries.Writer.TryWrite((context.GetValue<JsonElement>(0).Clone(), seq)))
    {
        Console.Error.WriteLine("Delivery queue full; restart from the saved cursor");
        Environment.Exit(1);
    }
    lastQueued = seq;
    return Task.CompletedTask;
});
client.On("delivery_error", context =>
{
    Console.Error.WriteLine($"Delivery error: {context.RawText}");
    return Task.CompletedTask;
});
client.OnError += (_, error) => Console.Error.WriteLine($"Connection error: {error}");
await client.ConnectAsync();

await foreach (var (ev, seq) in deliveries.Reader.ReadAllAsync())
{
    if (seq <= cursor) continue;
    await HandleEvent(ev);
    File.WriteAllText($"{CursorFile}.tmp", seq.ToString());
    File.Move($"{CursorFile}.tmp", CursorFile, overwrite: true);
    Interlocked.Exchange(ref cursor, seq);
}

static Task HandleEvent(JsonElement ev)
{
    Console.WriteLine($"{ev.GetProperty("type")} {ev.GetProperty("id")}"); // Your idempotent application handler.
    return Task.CompletedTask;
}

Run it from the directory where its cursor should be stored:

dotnet run

Client-specific recovery limits

These are examples, not a shared SDK. Initial connection errors and terminal server disconnects do not have identical retry behavior in every library. Read the connection error, correct the credential or project, and restart when needed. A negative subscribe acknowledgement is also an application error, not a reason to keep advancing the cursor. Rust’s emit_with_ack expires the callback after its timeout without calling it. The .NET example uses the library’s acknowledgement callback without adding an application watchdog. If a subscribe acknowledgement never arrives on an otherwise live connection, stop and restart that process; production workers should monitor subscription readiness. The Go example uses zishang520’s client because it supports Photon’s current protocol and required options. Do not substitute the legacy googollee/go-socket.io client: the tested v1.8.0-rc.1 does not support this endpoint’s Engine.IO v4, WebSocket-only transport, custom path, and header requirements.

Get started

Problems

Endpoints

Event Delivery WebSocket

Event schemas

Client examples

Node.js

Checkpointed examples

Python

Go

Rust

Java

.NET

Client-specific recovery limits

​Node.js

​Checkpointed examples

​Python

​Go

​Rust

​Java

​.NET

​Client-specific recovery limits

Node.js

Checkpointed examples

Python

Go

Rust

Java

.NET

Client-specific recovery limits