Events Reference

// Publish
client.SendEvent(ctx, kubemq.NewEvent().
    SetChannel("ch").SetBody([]byte("data")))

// Subscribe
client.SubscribeToEvents(ctx, "ch", "group",
    kubemq.WithOnEvent(handler),
    kubemq.WithOnError(errHandler))

// Stream Publish
streamCh := make(chan *kubemq.Event, 100)
resultCh := make(chan *kubemq.EventSendResult, 100)
go client.StreamEvents(ctx, streamCh, resultCh)

# Publish
client.send_event(EventMessage(channel="ch", body=b"data"))

# Subscribe
client.subscribe_to_events(
    EventsSubscription(channel="ch", group="group",
        on_receive_event_callback=handler,
        on_error_callback=err_handler),
    cancel=CancellationToken())

# Stream Publish
stream = client.open_events_stream()
stream.send(EventMessage(channel="ch", body=b"data"))

// Publish
await client.sendEvent({ channel: "ch", body: Buffer.from("data") });

// Subscribe
client.subscribeToEvents({
  channel: "ch", group: "group",
  onEvent: handler, onError: errHandler
});

// Stream Publish
const stream = client.createEventStream();
stream.send({ channel: "ch", body: Buffer.from("data") });

// Publish
client.sendEventsMessage(EventMessage.builder()
    .channel("ch").body("data".getBytes()).build());

// Subscribe
client.subscribeToEvents(EventsSubscription.builder()
    .channel("ch").group("group")
    .onReceiveEventCallback(handler)
    .onErrorCallback(errHandler).build());

// Stream Publish
EventsStream stream = client.openEventsStream();
stream.send(EventMessage.builder()
    .channel("ch").body("data".getBytes()).build());

// Publish
await client.SendEventAsync(new EventMessage {
    Channel = "ch", Body = Encoding.UTF8.GetBytes("data") });

// Subscribe
await foreach (var msg in client.SubscribeToEventsAsync(
    new EventsSubscription { Channel = "ch", Group = "group" })) { }

// Stream Publish
var stream = client.OpenEventsStream();
await stream.SendAsync(new EventMessage {
    Channel = "ch", Body = Encoding.UTF8.GetBytes("data") });

// Publish
client.sendEvent(EventMessage(
    channel = "ch", body = "data".toByteArray()))

// Subscribe
client.subscribeToEvents(
    channel = "ch", group = "group",
    onEvent = handler, onError = errHandler)

// Stream Publish
val stream = client.openEventsStream()
stream.send(EventMessage(channel = "ch", body = "data".toByteArray()))

// Publish
kubemq::EventMessage event;
event.channel = "ch";
event.body = "data";
client.sendEvent(event);

// Subscribe
client.subscribeToEvents("ch", "group", handler, errHandler);

// Stream Publish
auto stream = client.openEventsStream();
stream.send(event);

// Publish
let event = EventBuilder::new()
    .channel("ch").body(b"data".to_vec()).build();
client.send_event(event).await?;

// Subscribe
let sub = client.subscribe_to_events("ch", "group",
    |event| Box::pin(async move { handle(event).await }),
    None).await?;

// Stream Publish
let mut stream = client.send_event_stream().await?;
stream.send(EventBuilder::new()
    .channel("ch").body(b"data".to_vec()).build()).await?;

# Publish
client.send_event(KubeMQ::PubSub::EventMessage.new(
  channel: "ch", body: "data"))

# Subscribe
sub = KubeMQ::PubSub::EventsSubscription.new(channel: "ch", group: "group")
client.subscribe_to_events(sub, cancellation_token: cancel,
  on_error: ->(e) { handle_error(e) }) { |event| handle(event) }

# Stream Publish
sender = client.create_events_sender
sender.publish(KubeMQ::PubSub::EventMessage.new(
  channel: "ch", body: "data"))

# Publish
event = KubeMQ.Event.new(channel: "ch", body: "data")
KubeMQ.Client.send_event(client, event)

# Subscribe
{:ok, sub} = KubeMQ.Client.subscribe_to_events(client, "ch",
  group: "group", on_event: fn event -> handle(event) end)

# Stream Publish
{:ok, handle} = KubeMQ.Client.send_event_stream(client)
KubeMQ.EventStreamHandle.send(handle,
  KubeMQ.Event.new(channel: "ch", body: "data"))