Handle Slow Consumers

workCh := make(chan []byte, 1000)

go func() {
    for body := range workCh {
        processOrder(body) // slow work happens here
    }
}()

sub, err := client.SubscribeToEvents(ctx, "order-events", "",
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        select {
        case workCh <- event.Body:
        default:
            log.Println("Local buffer full, dropping event")
        }
    }),
    kubemq.WithOnError(func(err error) {
        log.Println("Error:", err)
    }),
)

import queue
import threading

work_queue = queue.Queue(maxsize=1000)

def worker():
    while True:
        body = work_queue.get()
        process_order(body)  # slow work happens here
        work_queue.task_done()

threading.Thread(target=worker, daemon=True).start()

def on_event(event):
    try:
        work_queue.put_nowait(event.body)
    except queue.Full:
        print("Local buffer full, dropping event")

client.subscribe_to_events(
    subscription=EventsSubscription(
        channel="order-events",
        on_receive_event_callback=on_event,
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

const workQueue = [];
const MAX_BUFFER = 1000;

setInterval(() => {
  while (workQueue.length > 0) {
    const body = workQueue.shift();
    processOrder(body); // slow work
  }
}, 10);

client.subscribeToEvents({
  channel: "order-events",
  onEvent: (msg) => {
    if (workQueue.length >= MAX_BUFFER) {
      console.warn("Local buffer full, dropping event");
      return;
    }
    workQueue.push(msg.body);
  },
  onError: (err) => console.error("Error:", err.message),
});

ExecutorService executor = Executors.newFixedThreadPool(4);
BlockingQueue<byte[]> workQueue = new LinkedBlockingQueue<>(1000);

executor.submit(() -> {
    while (true) {
        byte[] body = workQueue.take();
        processOrder(body); // slow work
    }
});

client.subscribeToEvents(EventsSubscription.builder()
    .channel("order-events")
    .onReceiveEventCallback(event -> {
        if (!workQueue.offer(event.getBody())) {
            System.err.println("Local buffer full, dropping event");
        }
    })
    .onErrorCallback(err ->
        System.err.println("Error: " + err.getMessage()))
    .build());

var workQueue = Channel.CreateBounded<byte[]>(1000);

_ = Task.Run(async () =>
{
    await foreach (var body in workQueue.Reader.ReadAllAsync())
    {
        ProcessOrder(body); // slow work
    }
});

await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "order-events" }))
{
    if (!workQueue.Writer.TryWrite(msg.Body.ToArray()))
    {
        Console.Error.WriteLine("Local buffer full, dropping event");
    }
}

val workQueue = LinkedBlockingQueue<ByteArray>(1000)

thread(isDaemon = true) {
    while (true) {
        val body = workQueue.take()
        processOrder(body) // slow work
    }
}

client.subscribeToEvents(
    channel = "order-events",
    onEvent = { event ->
        if (!workQueue.offer(event.body)) {
            System.err.println("Local buffer full, dropping event")
        }
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

std::queue<std::string> workQueue;
std::mutex queueMutex;
const size_t MAX_BUFFER = 1000;

std::thread worker([&]() {
    while (true) {
        std::string body;
        {
            std::lock_guard<std::mutex> lock(queueMutex);
            if (workQueue.empty()) continue;
            body = workQueue.front();
            workQueue.pop();
        }
        processOrder(body); // slow work
    }
});

client.subscribeToEvents("order-events", "",
    [&](const kubemq::Event& event) {
        std::lock_guard<std::mutex> lock(queueMutex);
        if (workQueue.size() >= MAX_BUFFER) {
            std::cerr << "Local buffer full, dropping event" << std::endl;
            return;
        }
        workQueue.push(event.body);
    },
    [](const std::string& err) {
        std::cerr << "Error: " << err << std::endl;
    }
);

use kubemq::prelude::*;
use kubemq::Subscription;
use tokio::sync::mpsc;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    // Bounded local buffer; a background task does the slow work
    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(1000);

    tokio::spawn(async move {
        while let Some(body) = rx.recv().await {
            process_order(body).await; // slow work happens here
        }
    });

    let sub: Subscription = client
        .subscribe_to_events(
            "order-events",
            "",
            move |event| {
                let tx = tx.clone();
                Box::pin(async move {
                    // try_send returns immediately; never block the callback
                    if tx.try_send(event.body).is_err() {
                        eprintln!("Local buffer full, dropping event");
                    }
                })
            },
            None,
        )
        .await?;

    tokio::signal::ctrl_c().await.ok();
    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: "localhost:50000", client_id: "order-subscriber")
cancel = KubeMQ::CancellationToken.new

# Bounded local buffer drained by a background worker thread
work_queue = SizedQueue.new(1000)

Thread.new do
  loop do
    body = work_queue.pop
    process_order(body) # slow work happens here
  end
end

sub = KubeMQ::PubSub::EventsSubscription.new(channel: "order-events")
client.subscribe_to_events(sub, cancellation_token: cancel,
                           on_error: ->(e) { warn "Error: #{e.message}" }) do |event|
  # push(..., true) is non-blocking; raises when the buffer is full
  begin
    work_queue.push(event.body, true)
  rescue ThreadError
    warn "Local buffer full, dropping event"
  end
end

sleep
cancel.cancel
client.close

{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "order-subscriber")

# Spawn a worker process; the callback only hands work off to it
worker =
  spawn(fn ->
    Stream.repeatedly(fn ->
      receive do
        {:work, body} -> process_order(body) # slow work happens here
      end
    end)
    |> Stream.run()
  end)

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events(client, "order-events",
    on_event: fn event ->
      # Guard against an unbounded mailbox; drop when the worker falls behind
      {:message_queue_len, len} = Process.info(worker, :message_queue_len)

      if len >= 1000 do
        IO.puts("Local buffer full, dropping event")
      else
        send(worker, {:work, event.body})
      end
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

Process.sleep(:infinity)
KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)