Cancel a crawl subscription

A crawl event subscription has no built-in maximum duration. An idle subscription may remain open indefinitely. The ordinary request timeout does not bound the stream: Python disables its read timeout, Go removes the HTTP client and per-request timeout, and TypeScript relies on the caller’s abort signal. Set a caller deadline when your application needs a bounded subscription.

The examples below subscribe to an existing crawl job; both tiers serve the stream. Set XBERG_BASE_URL, XBERG_API_KEY, and XBERG_CRAWL_JOB_ID. They default to a 30-second caller deadline; XBERG_STREAM_TIMEOUT_SECONDS overrides it. Their ordinary request timeout is deliberately shorter, demonstrating that it does not limit an idle stream.

Python

Use the async client to cancel even while the next event is blocked on network I/O. asyncio.wait_for cancels the consumer task when its deadline expires, and aclosing closes the generator when consumption ends. This works on Python 3.10 and later.

Python

import asyncio
import os
from contextlib import aclosing

from xberg_io_sdk import AsyncXbergClient


async def main():
    async with AsyncXbergClient(
        api_key=os.environ["XBERG_API_KEY"],
        base_url=os.environ["XBERG_BASE_URL"],
        target="enterprise",
        timeout=0.5,
    ) as client:

        async def consume():
            async with aclosing(client.stream_crawl_events(os.environ["XBERG_CRAWL_JOB_ID"])) as events:
                async for event in events:
                    print(event.kind)

        try:
            await asyncio.wait_for(consume(), timeout=float(os.environ.get("XBERG_STREAM_TIMEOUT_SECONDS", "30")))
        except asyncio.TimeoutError:
            print("Subscription canceled")


asyncio.run(main())

For synchronous consumption, wrap the generator in contextlib.closing when breaking early. Closing it after an event releases the response, but a deadline checked inside the loop cannot interrupt an idle read. Use the async pattern above when cancellation must work during idle periods.

TypeScript

Pass an AbortSignal to streamCrawlEvents. Aborting it interrupts an idle read and releases the response body. Breaking out of the loop also closes the response. Check the signal when handling cancellation so unrelated connection or protocol errors still propagate.

TypeScript

import { XbergClient } from "@xberg-io/sdk";

const baseUrl = process.env.XBERG_BASE_URL;
if (!baseUrl) throw new Error("Set XBERG_BASE_URL to your deployment URL");
const client = new XbergClient({
  apiKey: process.env.XBERG_API_KEY,
  baseUrl,
  target: "enterprise",
  timeoutMs: 500,
});
const crawlJobId = process.env.XBERG_CRAWL_JOB_ID;
if (!crawlJobId) throw new Error("Set XBERG_CRAWL_JOB_ID to an existing crawl job");
const signal = AbortSignal.timeout(Number(process.env.XBERG_STREAM_TIMEOUT_SECONDS ?? "30") * 1000);
try {
  for await (const event of client.streamCrawlEvents(crawlJobId, { signal })) {
    console.log(event.kind);
  }
} catch (error) {
  if (!signal.aborted) throw error;
}
if (signal.aborted) console.log("Subscription canceled");

Go

Pass a context with a deadline and defer its cancellation. The context interrupts the subscription while idle; breaking out of the iterator also closes the response body. Preserve errors that are unrelated to the caller deadline.

Go

package main

import (
  "context"
  "errors"
  "fmt"
  "os"
  "strconv"
  "time"

  xberg "github.com/xberg-io/sdks/packages/go"
)

func run() error {
  client, err := xberg.New(xberg.WithAPIKey(os.Getenv("XBERG_API_KEY")), xberg.WithBaseURL(os.Getenv("XBERG_BASE_URL")), xberg.WithTarget(xberg.TargetEnterprise), xberg.WithTimeout(500*time.Millisecond))
  if err != nil {
    return err
  }
  duration := 30.0
  if value := os.Getenv("XBERG_STREAM_TIMEOUT_SECONDS"); value != "" {
    duration, err = strconv.ParseFloat(value, 64)
    if err != nil {
      return err
    }
  }
  subscriptionContext, cancel := context.WithTimeout(context.Background(), time.Duration(duration*float64(time.Second)))
  defer cancel()
  events, err := client.StreamCrawlEvents(subscriptionContext, os.Getenv("XBERG_CRAWL_JOB_ID"))
  if err != nil {
    return err
  }
  for event, err := range events {
    if err != nil {
      if !errors.Is(subscriptionContext.Err(), context.DeadlineExceeded) {
        return err
      }
      break
    }
    fmt.Println(event.Kind)
  }
  if errors.Is(subscriptionContext.Err(), context.DeadlineExceeded) {
    fmt.Println("Subscription canceled")
  }
  return nil
}

func main() {
  if err := run(); err != nil {
    fmt.Fprintln(os.Stderr, err)
    os.Exit(1)
  }
}

Stopping a subscription closes that client’s connection. It does not cancel the crawl job itself. Other subscribers can continue to receive events, and a later subscription is a new connection.