NexJob.Trigger.GooglePubSub 6.1.0

NexJob.Trigger.GooglePubSub

Google Cloud Pub/Sub trigger for NexJob. Receives messages from a Pub/Sub subscription and automatically enqueues them as NexJob jobs.

Installation

dotnet add package NexJob.Trigger.GooglePubSub

Usage

Subscription Setup

using NexJob.Trigger.GooglePubSub;

builder.Services.AddNexJobGooglePubSubTrigger(options =>
{
    options.ProjectId = "my-gcp-project";
    options.SubscriptionId = "my-subscription";
    options.TargetQueue = "default";
});

Local Development with Pub/Sub Emulator

builder.Services.AddNexJobGooglePubSubTrigger(options =>
{
    options.ProjectId = "local-project";
    options.SubscriptionId = "local-subscription";
    options.EmulatorHost = "localhost:8085";
});

With EmulatorHost set, the trigger connects without TLS and without credentials. The repository tests the trigger against the real emulator image (gcr.io/google.com/cloudsdktool/cloud-sdk:emulators).

Strongly-Typed Consumer Registration

builder.Services.AddNexJobGooglePubSubTrigger<ProcessPubSubJob>(options =>
{
    options.ProjectId = "my-gcp-project";
    options.SubscriptionId = "my-subscription";
});

Operational Visibility

This trigger registers with IListenerRegistry to report live connection states (Starting, Listening, Faulted, Stopped) directly to the Dashboard /listeners page and the Cluster Pipeline Topology Map.

Configuration Options

Message Contract

The trigger expects messages with the following properties:

• Data: The job input payload as a UTF-8 string (JSON).
• Attributes:
  • nexjob.job_type: Assembly-qualified name of the job implementation type. Optional when the subscriber sets options.JobType or registers the trigger with AddNexJobGooglePubSubTrigger<TJob>(); the attribute wins when both are present. A message with neither can never become a job: it is nacked after a 2 second pause (so it is not redelivered in a tight loop) and logged as an error. Pub/Sub delivers a nacked message again, so give the subscription a dead-letter policy (a dead-letter topic with a maximum number of delivery attempts) to park such messages instead of redelivering them forever.
  • traceparent: W3C traceparent header for distributed tracing (optional).

Broker Guarantees

This trigger satisfies all canonical NexJob trigger guarantees:

1. At-least-once delivery — messages are delivered via pull subscription and acknowledged only after successful enqueue.
2. Explicit Ack / Nack — returns SubscriberClient.Reply.Ack strictly after IScheduler.EnqueueAsync completes. On enqueue failure, returns SubscriberClient.Reply.Nack to allow reprocessing.
3. Trace propagation — extracts W3C traceparent from message attributes and assigns it to JobRecord.TraceParent.
4. Idempotency — uses native Pub/Sub MessageId as the JobRecord.IdempotencyKey.
5. Graceful shutdown — respects cancellation token during host shutdown, nacking any in-flight messages so they can be picked up after restart.

Publishing Example

var publisher = await PublisherClient.CreateAsync(TopicName.FromProjectTopic("my-gcp-project", "my-topic"));

var message = new PubsubMessage
{
    Data = ByteString.CopyFromUtf8(jsonPayload),
    Attributes =
    {
        ["nexjob.job_type"] = typeof(OrderProcessingJob).AssemblyQualifiedName,
        ["traceparent"] = Activity.Current?.Id ?? string.Empty,
    },
};

await publisher.PublishAsync(message);