cloudevent

CloudEvent Module

NATS JetStream-backed job runtime for protobuf-defined CloudEvents. Handlers and publishers are wired from service/method options in .proto files and generated by protoc-gen-go-cloudevent2.

Features

• JetStream jobs: Streams, consumers, retries, and concurrency from YAML config
• Protobuf-first: Job name and subject/topic come from proto extensions
• Codegen: Register*CloudEvent, *Publisher, and subject/job key constants
• Interceptors: Subscribe and publish middleware chains
• Delivery control: Reject, Redelivery, and ForceRetry for consumer semantics
• Typed results: Publish returns result.Result[*PubAckInfo]

Installation

go get github.com/pubgo/funk/v2/component/cloudevent
go install github.com/pubgo/funk/v2/cmds/protoc-gen-go-cloudevent2@latest

From this repository root:

make protobuf   # regenerates proto output

Protobuf layout

CloudEvent uses two proto packages on purpose:

Message types and extensions are split so application protos can import only the extension package (which re-exports the option message types) without pulling extension definitions into every generated API surface.

Annotate a service

syntax = "proto3";

import "cloudeventoption/options.proto";

service GidInnerService {
  option (lava.cloudevent.job) = { name: "gid" };

  rpc ProxyExec(ProxyExecReq) returns (google.protobuf.Empty) {
    option (lava.cloudevent.subject) = { name: "gid.proxy.exec" };
  }
}

• job.name must match a consumer group key in YAML config (consumers.<job>).
• subject.name must match a subject entry under that consumer.

Codegen

Add to your protoc invocation (see root protobuf.yaml):

--go-cloudevent2_out=paths=source_relative:.

Generated artifacts per annotated service include:

• <Service>CloudEventJobKey — job name constant
• <Method>CloudEventSubjectKey — subject/topic constant
• <Service>CloudEvent — handler struct with On<Method> fields
• Register<Service>CloudEvent(client, event, opts...) — registers non-nil handlers
• <Service>Publisher — { Client, Opt, Interceptors } with Push<Method>Event methods

Configuration

Jobs are declared in YAML (see config.yaml for shape):

jobs:
  streams:
    gid:
      storage: "file"
      subjects: ["gid.>"]
  consumers:
    gid:
      - consumer: "test:gid"
        stream: "gid"
        subjects: "gid.proxy.exec"
        job:
          timeout: "1m"
          max_retries: 10

Load config and create a client with your NATS JetStream connection. Subject names in config are prefixed at runtime (DefaultPrefix, default acj).

Defaults (overridable per job):

Quick start

Register handlers (generated)

RegisterGidInnerServiceCloudEvent(jobCli, GidInnerServiceCloudEvent{
    OnProxyExec: func(ctx context.Context, req *gidpb.ProxyExecReq) error {
        evt := cloudevent.GetContext(ctx)
        // evt.Subject, evt.NumDelivered, evt.Config, ...
        return nil
    },
})

Register handlers (manual)

cloudevent.RegisterJobHandler(jobCli, "gid", "gid.proxy.exec",
    func(ctx context.Context, req *gidpb.ProxyExecReq) error { return nil },
    cloudevent.ProtoRegisterOpts(registerOpts...),
    cloudevent.WithSubInterceptors(myInterceptor),
)

Publish (generated publisher)

pub := GidInnerServicePublisher{
    Client: jobCli,
    Opt:    cloudevent.ProtoPubOpts(defaultPushOpts...),
}
ack := pub.PushProxyExecEvent(ctx, req).Unwrap()

Publish (direct)

ack := cloudevent.Publish(jobCli, ctx, subjectKey, req, interceptors, opts...).Unwrap()

Publish merges PubOpt values, sets default content_type to application/json, and fills sender from build metadata when omitted.

Interceptors

Subscribe (SubInterceptor): wrap the handler invocation. Attach per registration via WithSubInterceptors or RegisterOpt.

Publish (PubInterceptor): wrap the publish call. Pass on Client.Publish / generated Publisher.Interceptors.

Both follow a func(next ...) ... middleware pattern.

Delivery errors

Return these from handlers to control JetStream acknowledgment:

Ordinary returned errors trigger the configured retry/backoff policy until max_retries is exhausted.

Context

GetContext(ctx) returns delivery metadata: subject, stream, consumer, delivery counts, timestamp, HTTP-style headers, and resolved JobEventConfig. Replaces the old GetEventContext name.

Migration from pre-v2 funk cloudevent

Breaking changes when moving from the previous funk layout:

Checklist for downstream services:

1. Update proto imports to cloudeventoption/options.proto.
2. Regenerate with protoc-gen-go-cloudevent2 (not protoc-gen-go-cloudevent).
3. Replace handler registration with generated Register*CloudEvent or pass RegisterOpt.
4. Replace publish calls with *Publisher or Publish + result unwrapping.
5. Rename GetEventContext → GetContext.
6. Align YAML job/subject names with proto job / subject options.

Example

Runnable end-to-end demo (NATS + JetStream required):

docker run --rm -p 4222:4222 nats:latest -js
go run ./component/cloudevent/example

Source layout:

• example/main.go — wires NATS, YAML config, generated register/publish APIs
• example/demopb/demo.proto — sample service annotations
• example/config.yaml — stream/consumer config matching the proto subjects

Package-level godoc examples live in example_test.go.

References

• CloudEvents Go SDK (design inspiration)
• NATS JetStream