API Reference#

This page provides detailed reference information for both the Event Aggregation Service (Publisher API) and Event Consumer Service (Consumer API).

API Types#

Both services provide dual APIs:

gRPC API: Protocol Buffer-based, high-performance, strongly-typed
REST API: HTTP/JSON-based, easier for testing and web clients

Use gRPC for production microservices and REST for web clients or when easier HTTP tooling is preferred.

Event Aggregation Service (Publisher API)#

The Event Aggregation Service accepts events from publishers and routes them to consumers.

Base URLs#

gRPC: your-aggregation-service.example.com:50051
REST: https://your-aggregation-service.example.com

Authentication#

Include a bearer token in all requests:

gRPC:

metadata = [('authorization', 'Bearer YOUR_TOKEN')]

REST:

Authorization: Bearer YOUR_TOKEN

Publisher gRPC API#

Service Definition#

service EventPublishingService {
  rpc PublishEvent(PublishEventRequest) returns (PublishEventResponse);
  rpc BatchPublishEvents(BatchPublishEventsRequest) returns (BatchPublishEventsResponse);
}

PublishEvent#

Publishes a single event.

Request: PublishEventRequest

message PublishEventRequest {
  Event event = 1;
}

message Event {
  string event_type = 1;
  google.protobuf.Struct message = 2;
  google.protobuf.Timestamp occurred_at = 3;
  EventResource resource = 4;  // Optional
}

message EventResource {
  string resource_id = 1;
}

Response: PublishEventResponse

message PublishEventResponse {
  PublishingResult result = 1;
}

message PublishingResult {
  Event event = 1;
  bool success = 2;
  string failure_reason = 3;  // Only set if success is false
}

Errors:

gRPC Status Code	Meaning
`UNAUTHENTICATED`	Invalid or expired authentication token
`PERMISSION_DENIED`	Not authorized to publish this event type
`INVALID_ARGUMENT`	Invalid event structure
`RESOURCE_EXHAUSTED`	Service temporarily overloaded - retry with backoff
`INTERNAL`	Server error

Example:

from nvidia.omniverse.notifications.publisher.v1beta import event_publisher_pb2

# Create event
event = event_publisher_pb2.Event(
    event_type='myapp.user.created',
    message=message_struct,
    occurred_at=timestamp,
    resource=event_publisher_pb2.EventResource(
        resource_id='/users/123'
    )
)

# Publish
request = event_publisher_pb2.PublishEventRequest(event=event)
response = stub.PublishEvent(request, metadata=metadata)

if response.result.success:
    print("Published successfully")

BatchPublishEvents#

Publishes multiple events in parallel.

Request: BatchPublishEventsRequest

message BatchPublishEventsRequest {
  repeated Event events = 1;
}

Response: BatchPublishEventsResponse

message BatchPublishEventsResponse {
  repeated PublishingResult results = 1;
}

Note: Each event in the batch has its own result. Some may succeed while others fail.

Example:

# Create multiple events
events = [event1, event2, event3]

# Batch publish
request = event_publisher_pb2.BatchPublishEventsRequest()
for event in events:
    request.events.append(event)
response = stub.BatchPublishEvents(request, metadata=metadata)

# Check individual results
for idx, result in enumerate(response.results):
    if result.success:
        print(f"Event {idx} published")
    else:
        print(f"Event {idx} failed: {result.failure_reason}")

Publisher REST API#

Endpoints#

POST /api/v1beta/events#

Publishes a single event.

Request Body:

{
  "event": {
    "event_type": "string",
    "message": {},
    "occurred_at": "2024-10-16T14:30:00Z",
    "resource": {
      "resource_id": "/path/to/resource"
    }
  }
}

Response (200 OK):

{
  "result": {
    "event": { /* echo of event */ },
    "success": true
  }
}

Response (200 OK, but publish failed):

{
  "result": {
    "event": { /* echo of event */ },
    "success": false,
    "failure_reason": "Unable to connect to message broker"
  }
}

Error Responses:

Status Code	Description
400	Invalid request format
401	Unauthenticated
403	Permission denied
422	Validation error
503	Service unavailable - retry

Example:

curl -X POST https://your-aggregation-service.example.com/api/v1beta/events \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "event": {
      "event_type": "myapp.user.created",
      "message": {
        "user_id": "123",
        "username": "john_doe"
      },
      "occurred_at": "2024-10-16T14:30:00Z",
      "resource": {
        "resource_id": "/users/123"
      }
    }
  }'

POST /api/v1beta/events/batch#

Publishes multiple events in parallel.

Request Body:

{
  "events": [
    {
      "event_type": "string",
      "message": {},
      "occurred_at": "2024-10-16T14:30:00Z",
      "resource": {
        "resource_id": "/path/to/resource"
      }
    }
  ]
}

Response (200 OK):

{
  "results": [
    {
      "event": { /* event 1 */ },
      "success": true
    },
    {
      "event": { /* event 2 */ },
      "success": false,
      "failure_reason": "Publish failed"
    }
  ]
}

Example:

curl -X POST https://your-aggregation-service.example.com/api/v1beta/events/batch \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "events": [
      {
        "event_type": "myapp.file.uploaded",
        "message": {"file": "doc1.pdf"},
        "occurred_at": "2024-10-16T14:30:00Z"
      },
      {
        "event_type": "myapp.file.uploaded",
        "message": {"file": "doc2.pdf"},
        "occurred_at": "2024-10-16T14:30:01Z"
      }
    ]
  }'

Event Consumer Service (Consumer API)#

The Event Consumer Service allows clients to stream events with filtering.

Base URLs#

gRPC: your-consumer-service.example.com:50052
REST: https://your-consumer-service.example.com

Consumer gRPC API#

Service Definition#

service EventConsumerService {
  rpc ConsumeNonDurableEvents(stream ConsumeNonDurableEventsRequest) 
      returns (stream ConsumeNonDurableEventsResponse);
  
  // Deprecated: use ConsumeDurableEventsWithAcks instead.
  rpc ConsumeDurableEvents(ConsumeDurableEventsRequest) 
      returns (stream ConsumeDurableEventsResponse);
  
  rpc ConsumeDurableEventsWithAcks(stream ConsumeDurableEventsWithAcksRequest) 
      returns (stream ConsumeDurableEventsWithAcksResponse);
  
  rpc CreateDurableQueue(CreateDurableQueueRequest) 
      returns (CreateDurableQueueResponse);
  
  rpc UpdateDurableQueue(UpdateDurableQueueRequest) 
      returns (UpdateDurableQueueResponse);
  
  rpc DeleteDurableQueue(DeleteDurableQueueRequest) 
      returns (DeleteDurableQueueResponse);
  
  rpc ManageDeadLetteredEvents(stream ManageDeadLetteredEventsRequest) 
      returns (stream ManageDeadLetteredEventsResponse);
}

ConsumeNonDurableEvents#

Streams events using a non-durable queue (bidirectional streaming).

Request: ConsumeNonDurableEventsRequest (stream)

message ConsumeNonDurableEventsRequest {
  repeated FilterGroup filter_groups = 1;
  optional string reconnect_token = 2;
  repeated FilterGroup previous_filter_groups = 3;  // Required when updating filters
}

message FilterGroup {
  string event_type = 1;
  repeated ResourceFilter filters = 2;
}

message ResourceFilter {
  enum FilterType {
    FILTER_TYPE_UNSPECIFIED = 0;
    FILTER_TYPE_EQ = 1;
    FILTER_TYPE_STARTS_WITH_LAZY = 2;
    FILTER_TYPE_STARTS_WITH_GREEDY = 3;
  }
  
  FilterType filter_type = 1;
  string resource_id = 2;
}

Response: ConsumeNonDurableEventsResponse (stream)

message ConsumeNonDurableEventsResponse {
  repeated Event events = 1;
  string reconnect_token = 2;
}

message Event {
  string event_type = 1;
  string principal_identity = 2;
  google.protobuf.Timestamp occurred_at = 3;
  google.protobuf.Timestamp published_at = 4;
  google.protobuf.Struct message = 5;
}

Example:

# Create filter
filter_group = event_consumer_pb2.FilterGroup(
    event_type='myapp.user.created'
)
# No filters added - receive all events of this type

# Request stream
def request_stream():
    request = event_consumer_pb2.ConsumeNonDurableEventsRequest()
    request.filter_groups.append(filter_group)
    yield request

# Consume
responses = stub.ConsumeNonDurableEvents(request_stream(), metadata=metadata)

for response in responses:
    reconnect_token = response.reconnect_token  # Save for reconnection
    
    for event in response.events:
        print(f"Event: {event.event_type}")
        print(f"Message: {event.message}")

ConsumeDurableEvents (Deprecated)#

Deprecated: Use ConsumeDurableEventsWithAcks instead. This RPC acknowledges messages automatically after a short delay, which has a window for message loss on disconnect. It will be removed in a future release.

Streams events from a durable queue (server streaming). Messages are automatically acknowledged by the server after a short delay.

Request: ConsumeDurableEventsRequest

message ConsumeDurableEventsRequest {
  string queue_id = 1;
}

Response: ConsumeDurableEventsResponse (stream)

message ConsumeDurableEventsResponse {
  repeated Event events = 1;
}

Example:

request = event_consumer_pb2.ConsumeDurableEventsRequest(
    queue_id='saved-queue-id'
)

responses = stub.ConsumeDurableEvents(request, metadata=metadata)

for response in responses:
    for event in response.events:
        process_event(event)

ConsumeDurableEventsWithAcks#

Streams events from a durable queue with explicit client acknowledgment (bidirectional streaming). The client must acknowledge received messages by sending their delivery tags back in subsequent requests. The server redelivers unacknowledged messages on reconnect, up to a configurable maximum number of attempts. Rather than redelivering a message indefinitely, the server moves it to a dead-letter queue once it exceeds the redelivery limit, outlives the queue’s configured message TTL, or is evicted when the queue reaches its configured maximum length.

Request: ConsumeDurableEventsWithAcksRequest (stream)

message ConsumeDurableEventsWithAcksRequest {
  string queue_id = 1;                  // Required on first request; ignored after
  repeated int64 ack_delivery_tags = 2; // Delivery tags to acknowledge
}

The first message in the stream must include queue_id to identify the durable queue. Subsequent messages should include ack_delivery_tags to acknowledge receipt of previously delivered events. Acknowledge promptly: an unacknowledged message is redelivered on reconnect a bounded number of times and is then moved to a dead-letter queue rather than redelivered indefinitely.

A message also leaves the durable queue for the dead-letter queue when it outlives the queue’s configured message TTL or is evicted because the queue has reached its configured maximum length. The dead-letter queue is the destination for these poison, expired, and overflowed messages. See the bounded redelivery and dead-letter queue explanation in the consuming-events guide for the fuller treatment.

Response: ConsumeDurableEventsWithAcksResponse (stream)

message ConsumeDurableEventsWithAcksResponse {
  repeated Event events = 1;
  int64 delivery_tag = 2;  // Tag to send back to acknowledge this message
}

Example:

import queue

ack_queue = queue.Queue()

def request_stream(queue_id):
    # First request: identify the durable queue
    yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
        queue_id=queue_id
    )
    # Subsequent requests: send acknowledgments
    while True:
        tags = ack_queue.get()
        if tags is None:
            break
        yield event_consumer_pb2.ConsumeDurableEventsWithAcksRequest(
            ack_delivery_tags=tags
        )

responses = stub.ConsumeDurableEventsWithAcks(
    request_stream('saved-queue-id'), metadata=metadata
)

for response in responses:
    for event in response.events:
        process_event(event)
    # Acknowledge after processing
    ack_queue.put([response.delivery_tag])

CreateDurableQueue#

Creates a durable queue for persistent event storage.

This operation has ensure-exists semantics when you supply a queue_id: it creates the queue if no queue with that id exists, and is a no-op if a queue with that id already exists. The created flag in the response tells you which path the call took.

Request: CreateDurableQueueRequest

message CreateDurableQueueRequest {
  repeated FilterGroup filter_groups = 1;
  string queue_id = 2;
}

Field	Type	Description
`filter_groups`	`repeated FilterGroup`	Filters to apply to events. If omitted or empty, all events for an event type are added to the queue.
`queue_id`	`string`	Optional, client-specified queue identifier. An empty string means not supplied — the server generates one. When supplied, it must match `^[A-Za-z0-9._-]{1,200}$`; an id that fails this rule is rejected with `INVALID_ARGUMENT`.

Response: CreateDurableQueueResponse

message CreateDurableQueueResponse {
  string queue_id = 1;  // Save this!
  bool created = 2;
}

Field	Type	Description
`queue_id`	`string`	The id of the durable queue. Persist this and send it on all `ConsumeDurableEvents` calls.
`created`	`bool`	`true` when this call created a new queue, `false` when an existing queue with that id was found and returned unchanged (a no-op).

Example:

filter_group = event_consumer_pb2.FilterGroup(
    event_type='myapp.file.uploaded'
)
resource_filter = event_consumer_pb2.ResourceFilter(
    filter_type=event_consumer_pb2.ResourceFilter.FILTER_TYPE_STARTS_WITH_GREEDY,
    resource_id='/uploads/'
)
filter_group.filters.append(resource_filter)

request = event_consumer_pb2.CreateDurableQueueRequest()
request.filter_groups.append(filter_group)

# Optional: supply your own queue_id for ensure-exists semantics.
# Omit it (leave it empty) to have the server generate one.
request.queue_id = 'myapp-uploads-queue'

response = stub.CreateDurableQueue(request, metadata=metadata)
queue_id = response.queue_id

if response.created:
    print(f'Created new durable queue {queue_id}')
else:
    print(f'Reusing existing durable queue {queue_id}')

# IMPORTANT: Save queue_id to configuration

UpdateDurableQueue#

Updates filters on an existing durable queue.

Request: UpdateDurableQueueRequest

message UpdateDurableQueueRequest {
  string queue_id = 1;
  repeated FilterGroup current_filter_groups = 2;
  repeated FilterGroup new_filter_groups = 3;
}

Response: UpdateDurableQueueResponse

message UpdateDurableQueueResponse {}

DeleteDurableQueue#

Deletes a durable queue.

Request: DeleteDurableQueueRequest

message DeleteDurableQueueRequest {
  string queue_id = 1;
}

Response: DeleteDurableQueueResponse

message DeleteDurableQueueResponse {}

Example:

request = event_consumer_pb2.DeleteDurableQueueRequest(
    queue_id=queue_id
)

stub.DeleteDurableQueue(request, metadata=metadata)

ManageDeadLetteredEvents#

Operator-facing, bidirectional streaming RPC for inspecting and triaging dead-lettered events across the deployment. The server streams dead-lettered messages with triage metadata; the client replies with per-message actions (ACTION_REPLAY, ACTION_DISCARD, ACTION_LEAVE) keyed by a session-scoped delivery tag; the server streams back a correlated outcome report for each action.

gRPC-only: There is no REST endpoint for this RPC.

Required permission: event-consumer-service:inspect-and-replay-all-dlq. A caller lacking it is denied with PERMISSION_DENIED.

This no-store bidirectional RPC is the permanent baseline DLQ-management surface; it requires no datastore beyond the message broker. It is the forward-compatible floor: future additive unary store-backed RPCs (list/filter/replay-by-id/stats) may be added later without changing this RPC. See Managing the Dead-Letter Queue (Operators) in the consuming-events guide.

Request: ManageDeadLetteredEventsRequest (stream)

message ManageDeadLetteredEventsRequest {
  enum Action {
    ACTION_UNSPECIFIED = 0;
    ACTION_REPLAY = 1;   // Republish to origin queue (confirm-gated, no-loss)
    ACTION_DISCARD = 2;  // Ack and drop from the dead-letter queue
    ACTION_LEAVE = 3;    // Leave in place for later triage
  }

  message ActionEntry {
    int64 delivery_tag = 1;  // Session-scoped tag of the message to act on
    Action action = 2;       // Action to take on that message
  }

  repeated ActionEntry actions = 1;  // Empty on the first (opening) message
}

The first message opens the stream and carries no actions (the dead-letter queue is global, so there is no queue_id). Subsequent messages carry one or more ActionEntry items, each pairing a session-scoped delivery_tag with the Action to take, so multiple dead-lettered messages can be actioned in a single request.

Request fields:

Field	Type	Description
`actions`	`repeated ActionEntry`	Per-delivery-tag triage actions. Empty on the opening message.
`ActionEntry.delivery_tag`	`int64`	Session-scoped tag of the dead-lettered message, as received in a `dead_lettered_event`. Valid only within the live stream; not persisted or reused across reconnects.
`ActionEntry.action`	`Action`	`ACTION_REPLAY`, `ACTION_DISCARD`, or `ACTION_LEAVE`.

Response: ManageDeadLetteredEventsResponse (stream)

Each response carries exactly one of two arms via a oneof payload: a dead_lettered_event (a browse item) or an action_outcome (a report correlated to a requested action by delivery_tag).

message ManageDeadLetteredEventsResponse {
  message DeadLetteredEvent {
    enum Reason {
      REASON_UNSPECIFIED = 0;
      REASON_DELIVERY_LIMIT = 1;  // Exceeded its redelivery limit
      REASON_EXPIRED = 2;         // Outlived the queue's configured message TTL
      REASON_MAXLEN = 3;          // Evicted when queue reached its maximum length
    }

    Reason reason = 1;
    string origin_queue = 2;                         // Origin durable queue name
    int64 delivery_count = 3;                        // Broker delivery attempts
    google.protobuf.Timestamp dead_lettered_at = 4;  // When it was dead-lettered
    Event event = 5;                                 // event_type, principal_identity, message
    string resource_id = 6;                          // Resource id of the event
    int64 delivery_tag = 7;                          // Session-scoped tag to act on
  }

  message ActionOutcome {
    enum Outcome {
      OUTCOME_UNSPECIFIED = 0;
      OUTCOME_REPLAYED = 1;              // Republished to origin queue; DLQ copy removed
      OUTCOME_DISCARDED = 2;             // Acked and dropped from the dead-letter queue
      OUTCOME_LEFT = 3;                  // Left in place for later triage
      OUTCOME_ORIGIN_QUEUE_DELETED = 4;  // Origin queue gone; DLQ copy left in place (no-loss)
      OUTCOME_REPLAY_FAILED = 5;         // Replay unconfirmed; DLQ copy left in place (no-loss)
    }

    int64 delivery_tag = 1;  // Matches the tag the client sent
    Outcome outcome = 2;
    string detail = 3;       // Human-readable context (e.g. missing queue name)
  }

  oneof payload {
    DeadLetteredEvent dead_lettered_event = 1;
    ActionOutcome action_outcome = 2;
  }
}

DeadLetteredEvent (browse item) fields:

Field	Type	Description
`reason`	`Reason`	Why the message was dead-lettered: `REASON_DELIVERY_LIMIT`, `REASON_EXPIRED`, or `REASON_MAXLEN`.
`origin_queue`	`string`	Name of the durable queue the message was dead-lettered from.
`delivery_count`	`int64`	Broker delivery attempts before dead-lettering.
`dead_lettered_at`	`Timestamp`	When the message was dead-lettered.
`event`	`Event`	The dead-lettered event (`event_type`, `principal_identity`, `message`).
`resource_id`	`string`	Resource id associated with the event.
`delivery_tag`	`int64`	Session-scoped tag the client sends back to act on the message. Live-stream-only.

ActionOutcome (report) fields:

Field	Type	Description
`delivery_tag`	`int64`	Session-scoped tag identifying the message the outcome pertains to.
`outcome`	`Outcome`	`OUTCOME_REPLAYED`, `OUTCOME_DISCARDED`, `OUTCOME_LEFT`, `OUTCOME_ORIGIN_QUEUE_DELETED`, or `OUTCOME_REPLAY_FAILED`.
`detail`	`string`	Human-readable context (e.g. the missing origin queue name).

Each action the operator sends produces exactly one correlated action_outcome, keyed by delivery_tag, so the operator can observe the result of every action. ACTION_REPLAY is confirm-gated and no-loss: the dead-letter copy is removed only after the publish to the origin queue is confirmed. A ACTION_REPLAY whose origin queue has been deleted does not lose the message — the server leaves the dead-letter copy in place (ACTION_LEAVE semantics) and reports an OUTCOME_ORIGIN_QUEUE_DELETED outcome to the operator rather than silently handling it; re-create the origin queue to obtain a viable replay target. A replay that cannot be confirmed reports OUTCOME_REPLAY_FAILED and likewise leaves the copy in place.

Tenant scope: ACTION_REPLAY returns a message to its exact origin queue (the queue name embeds the hashed principal id), so replay does not cross tenants. The permission is coarse and grants cross-tenant dead-letter visibility and action by design; re-consume of a replayed message still runs the existing per-event authorization check.

Consumer REST API#

Endpoints#

GET /api/v1beta/events/stream#

Streams events using a non-durable queue (Server-Sent Events).

Query Parameters:

Parameter	Type	Required	Description
`filter_groups`	string (JSON)	No	URL-encoded JSON array of FilterGroup objects
`previous_filter_groups`	string (JSON)	No	URL-encoded JSON array of the previous FilterGroup objects (required when updating filters)
`reconnect_token`	string	No	Token for reconnecting to same queue

FilterGroup Structure:

[
  {
    "event_type": "string",
    "filters": [
      {
        "filter_type": "starts_with_greedy",
        "resource_id": "/path/"
      }
    ]
  }
]

Filter Types:

eq: Exact match
starts_with_lazy: Shallow prefix match
starts_with_greedy: Deep prefix match
unspecified: Defaults to eq

Response (text/event-stream):

event: connected
id: my-queue-id
data: {"message":"Connection established","reconnect_token":"my-queue-id"}

event: event
data: {"event_type":"myapp.user.created","principal_identity":"user@example.com","occurred_at":"2024-10-16T14:30:00Z","published_at":"2024-10-16T14:30:01Z","message":{"user_id":"123"}}

The initial connected event includes the reconnect_token (which is the queue ID). Save this token for reconnection.

Example:

# Simple consumption
curl -N https://your-consumer-service.example.com/api/v1beta/events/stream \
  -H "Authorization: Bearer YOUR_TOKEN"

# With filtering
FILTERS='[{"event_type":"myapp.user.created","filters":[]}]'
curl -N https://your-consumer-service.example.com/api/v1beta/events/stream \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -G --data-urlencode "filter_groups=$FILTERS"

GET /api/v1beta/events/stream-durable#

Streams events from a durable queue (Server-Sent Events). Messages are acknowledged to the broker automatically by the server after a short delay. Clients do not need to acknowledge individual messages.

If your application requires explicit client-side acknowledgment of each message (e.g. to guarantee at-least-once processing), use the gRPC ConsumeDurableEventsWithAcks RPC instead. That RPC uses a bidirectional stream, which keeps the acknowledgment on the same connection and works correctly across multi-pod deployments.

Query Parameters:

Parameter	Type	Required	Description
`queue_id`	string	Yes	ID of the durable queue

Response (text/event-stream):

event: connected
id: my-queue-id
data: {"message":"Connection established","reconnect_token":"my-queue-id"}

event: event
data: {"event_type":"myapp.user.created",...}

event: event
data: {"event_type":"myapp.user.updated",...}

Example:

curl -N https://your-consumer-service.example.com/api/v1beta/events/stream-durable?queue_id=my-queue-id \
  -H "Authorization: Bearer YOUR_TOKEN"

POST /api/v1beta/queues/durable#

Creates a durable queue.

This endpoint has ensure-exists semantics when you supply queue_id: it creates the queue if no queue with that id exists, and is a no-op if a queue with that id already exists. The created field in the response tells you which path the call took. An invalid queue_id (one that fails the ^[A-Za-z0-9._-]{1,200}$ validation rule) is rejected with 422 Unprocessable Entity.

Request Body:

{
  "filter_groups": [
    {
      "event_type": "string",
      "filters": [
        {
          "filter_type": "starts_with_greedy",
          "resource_id": "/path/"
        }
      ]
    }
  ],
  "queue_id": "myapp-uploads-queue"
}

queue_id is optional and client-specified. Omit it or send an empty string to have the server generate one. When supplied, it must match ^[A-Za-z0-9._-]{1,200}$.

The HTTP status code signals which path the request took:

201 Created — the request created a new durable queue (created is true). This is the case when you supply a queue_id that does not yet exist.
200 OK — the request resolved to a queue that already existed (created is false), or you omitted queue_id and the server generated one (created is true).

Response (201 Created, created-new path):

{
  "queue_id": "durable-queue-abc123",
  "created": true
}

created is true when this call created a new queue and false when an existing queue with the supplied queue_id was found and returned unchanged (a no-op). Branch on the status code to tell the two apart.

Example:

curl -X POST https://your-consumer-service.example.com/api/v1beta/queues/durable \
  -H "Authorization: Bearer YOUR_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "filter_groups": [
      {
        "event_type": "myapp.file.uploaded",
        "filters": [
          {
            "filter_type": "starts_with_greedy",
            "resource_id": "/uploads/"
          }
        ]
      }
    ],
    "queue_id": "myapp-uploads-queue"
  }'

PATCH /api/v1beta/queues/durable#

Updates filters on a durable queue.

Request Body:

{
  "queue_id": "string",
  "current_filter_groups": [ /* array of FilterGroup */ ],
  "new_filter_groups": [ /* array of FilterGroup */ ]
}

Response (200 OK):

{
  "success": true
}

DELETE /api/v1beta/queues/durable#

Deletes a durable queue.

Query Parameters:

Parameter	Type	Required	Description
`queue_id`	string	Yes	ID of the queue to delete

Response (200 OK):

{
  "success": true
}

Example:

curl -X DELETE "https://your-consumer-service.example.com/api/v1beta/queues/durable?queue_id=my-queue-id" \
  -H "Authorization: Bearer YOUR_TOKEN"

Error Handling#

Common HTTP Status Codes#

Code	Meaning	Action
200	Success	Process response
400	Bad Request	Check request format
401	Unauthorized	Refresh authentication token
403	Forbidden	Check permissions
404	Not Found	Check queue_id or endpoint
422	Validation Error	Check request validation
500	Internal Server Error	Retry after delay, contact support
503	Service Unavailable	Retry with exponential backoff

Common gRPC Status Codes#

Code	Meaning	Action
OK	Success	Process response
INVALID_ARGUMENT	Bad request	Check request format
UNAUTHENTICATED	Auth failed	Refresh token
PERMISSION_DENIED	No permission	Check permissions
NOT_FOUND	Resource not found	Check identifiers
RESOURCE_EXHAUSTED	Service overloaded	Retry with backoff
UNAVAILABLE	Connection lost (e.g. load balancer idle timeout)	Reconnect automatically
INTERNAL	Server error	Retry after delay

Protocol Buffer Files#

The complete protocol buffer definitions are available in the released archive or protobuf registry.

Publisher Proto: nvidia/omniverse/notifications/publisher/v1beta/event_publisher.proto

Consumer Proto: nvidia/omniverse/notifications/consumer/v1beta/event_consumer.proto

OpenAPI Specifications#

Full OpenAPI specifications are available in the released archive

Rate Limits and Quotas#

Currently, there are no enforced rate limits at the application level. However:

The service may return RESOURCE_EXHAUSTED (gRPC) or 503 (HTTP) if temporarily overloaded
Implement exponential backoff when you receive these errors
For high-volume publishing, use batch publishing for better performance

Versioning#

The current API version is v1beta.

APIs are under the /api/v1beta/ path (REST) or v1beta package (gRPC)
Beta APIs may change, but we will maintain backward compatibility where possible
When the API reaches GA, it will be versioned as v1

Service Health#

Both services expose health check endpoints:

gRPC: Standard gRPC health checking protocol

REST:

GET /health - Service health status
Returns 200 if healthy

Metrics and Monitoring#

Both services expose OpenTelemetry metrics and traces when telemetry is enabled via configuration.

Next Steps#

Try the Getting Started Guide for hands-on examples
Learn about Publishing Events
Learn about Consuming Events
Understand Permissions & Authorization