Help us shape the future of Go! Take the 2026 Go Developer Survey

ocistreaming

package
v1.9.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 27, 2026 License: MIT Imports: 28 Imported by: 0

README

OCI Streaming

Parity grade: B · SDK github.com/oracle/oci-go-sdk/v65@v65.126.2 · last audited 2026-10-25 (2d5590dc6)

Coverage

Metric Value
PARITY entries audited 30 (19 ok, 11 partial)
Feature families 9 (8 ok, 1 partial)
Known gaps 9
Deferred items 0
Resource leaks unknown
Known gaps
  • The Kafka compatibility API (bootstrap servers, SASL and the Kafka wire protocol) is not implemented; kafkaSettings are stored and returned and bootstrapServers is whatever the caller set.
  • Create, delete and update responses report the transitional state but updates and compartment moves complete immediately; there is no UPDATING state and work requests only model CREATE, UPDATE and DELETE completion times.
  • The 2 MiB/s consumer throughput and per-group request-rate limits are not enforced; only the 1 MiB/s per-partition ingest, 1 MiB request and 256 byte key limits are.
  • Partition and stream counts per tenancy, partition limits per stream (1 to 1000 here) and the exact error code strings for failed PutMessages entries are unverified against the live service.
  • Error codes and statuses for a duplicate stream name (409 Conflict), deleting a non-empty pool (409 IncorrectState), a lost group reservation on commit (409 IncorrectState) and an unusable customEncryptionKey (400 InvalidParameter) are unverified.
  • Private stream pools only store their subnet and NSG settings; endpoints are not network-isolated and endpointFqdn is the addressed host.
  • Messages and consumer group state are held in memory and in snapshots, so large retained volumes grow the snapshot.
  • opc-retry-token idempotency, IAM policy checks, resource principals and instance principals are not applied; only API-key signatures are verified.
  • Tag values and defined-tag namespaces are stored verbatim, not validated against a tag namespace.

More

Documentation

Index

Constants

View Source 
const DefaultPort = 8152

DefaultPort is the dedicated listener's default TCP port.

View Source 
const (
	// Region is the fixed region baked into emulated OCIDs.
	Region = "us-ashburn-1"
)

Variables

View Source 
var ErrNilAppContext = errors.New("ocistreaming: nil app context")

ErrNilAppContext is returned when Init is called with a nil AppContext.

View Source 
var ErrSnapshotNull = errors.New("ocistreaming: null entry in snapshot")

ErrSnapshotNull is returned when a snapshot contains a null entry.

Functions

This section is empty.

Types

type APIError 

type APIError struct {
	Code    string
	Message string
	Status  int
}

APIError is an OCI error response: HTTP status plus the `code` and `message` body fields.

func (*APIError) Error 

func (e *APIError) Error() string

type ConfigProvider 

type ConfigProvider interface {
	GetOCIStreamingSettings() Settings
}

ConfigProvider extracts OCI Streaming configuration from the AppContext Config.

type CreateCursorIn 

type CreateCursorIn struct {
	Offset    *int64     `json:"offset"`
	Time      *time.Time `json:"time"`
	Partition string     `json:"partition"`
	Type      string     `json:"type"`
}

CreateCursorIn is the CreateCursor request body.

type CreateGroupCursorIn 

type CreateGroupCursorIn struct {
	Time         *time.Time `json:"time"`
	CommitOnGet  *bool      `json:"commitOnGet"`
	TimeoutInMs  *int       `json:"timeoutInMs"`
	Type         string     `json:"type"`
	GroupName    string     `json:"groupName"`
	InstanceName string     `json:"instanceName"`
}

CreateGroupCursorIn is the CreateGroupCursor request body.

type CreatePoolIn 

type CreatePoolIn struct {
	Kafka   *KafkaSettings   `json:"kafkaSettings"`
	Key     *kmsKeyIn        `json:"customEncryptionKeyDetails"`
	Private *PrivateEndpoint `json:"privateEndpointDetails"`
	Tags
	Name        string `json:"name"`
	Compartment string `json:"compartmentId"`
}

CreatePoolIn is the CreateStreamPool request body.

type CreateStreamIn 

type CreateStreamIn struct {
	Tags
	Partitions       *int   `json:"partitions"`
	RetentionInHours *int   `json:"retentionInHours"`
	Name             string `json:"name"`
	Compartment      string `json:"compartmentId"`
	PoolID           string `json:"streamPoolId"`
}

CreateStreamIn is the CreateStream request body.

type Group 

type Group struct {
	Instances map[string]time.Time `json:"instances"`
	Rotation  map[string]int       `json:"rotation"`
	StreamID  string               `json:"streamId"`
	Name      string               `json:"name"`
	Next      []int64              `json:"next"`
	Reserved  []Reservation        `json:"reserved"`
}

Group is a consumer group of a stream.

type Handler 

type Handler struct {
	Backend *InMemoryBackend

	Port        int
	EnforceAuth bool
	// contains filtered or unexported fields
}

Handler serves OCI Streaming: stream pools, streams, connect harnesses and the messages API.

func NewHandler 

func NewHandler(backend *InMemoryBackend) *Handler

NewHandler creates a Handler; callers override Port and EnforceAuth from Settings.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation returns the operation name for metrics.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource returns the addressed resource id for metrics.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations lists the supported operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler for all OCI Streaming requests.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority is irrelevant because RouteMatcher never matches.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all in-memory state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher never matches: OCI Streaming runs on its own listener.

func (*Handler) Shutdown 

func (h *Handler) Shutdown(ctx context.Context)

Shutdown stops the listener and the sweeper.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

func (*Handler) StartWorker 

func (h *Handler) StartWorker(ctx context.Context) error

StartWorker binds the dedicated listener synchronously so a port conflict is returned to the caller.

type Harness 

type Harness struct {
	TimeCreated   time.Time  `json:"timeCreated"`
	TransitionAt  *time.Time `json:"transitionAt,omitempty"`
	Tags          Tags       `json:"tags"`
	ID            string     `json:"id"`
	CompartmentID string     `json:"compartmentId"`
	Name          string     `json:"name"`
	State         string     `json:"state"`
	Etag          string     `json:"etag"`
	Seq           int64      `json:"seq"`
}

Harness is a Kafka Connect harness.

type InMemoryBackend 

type InMemoryBackend struct {
	Auth *ociauth.Registry
	// contains filtered or unexported fields
}

InMemoryBackend holds every pool, stream, harness, group and work request under one lock.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates an empty backend.

func (*InMemoryBackend) ChangeHarnessCompartment 

func (b *InMemoryBackend) ChangeHarnessCompartment(id, ifMatch, compartment string) (string, error)

ChangeHarnessCompartment moves a connect harness to another compartment.

func (*InMemoryBackend) ChangePoolCompartment 

func (b *InMemoryBackend) ChangePoolCompartment(id, ifMatch, compartment string) (string, error)

ChangePoolCompartment moves a pool to another compartment.

func (*InMemoryBackend) ChangeStreamCompartment 

func (b *InMemoryBackend) ChangeStreamCompartment(id, ifMatch, compartment string) (string, error)

ChangeStreamCompartment moves a stream to another compartment.

func (*InMemoryBackend) Compartments 

func (b *InMemoryBackend) Compartments() []string

Compartments lists the distinct compartments that hold pools or streams.

func (*InMemoryBackend) ConsumerCommit 

func (b *InMemoryBackend) ConsumerCommit(streamID, cursor string) (cursorWire, error)

ConsumerCommit commits the offsets a group cursor accumulated.

func (*InMemoryBackend) ConsumerHeartbeat 

func (b *InMemoryBackend) ConsumerHeartbeat(streamID, cursor string) (cursorWire, error)

ConsumerHeartbeat extends the instance's reservations.

func (*InMemoryBackend) CreateCursor 

func (b *InMemoryBackend) CreateCursor(streamID string, in CreateCursorIn) (cursorWire, error)

CreateCursor creates a single-partition cursor.

func (*InMemoryBackend) CreateGroupCursor 

func (b *InMemoryBackend) CreateGroupCursor(streamID string, in CreateGroupCursorIn) (cursorWire, error)

CreateGroupCursor joins an instance to a consumer group, creating the group on first use.

func (*InMemoryBackend) CreateHarness 

func (b *InMemoryBackend) CreateHarness(in harnessIn) (harnessWire, string, string, error)

CreateHarness creates a connect harness.

func (*InMemoryBackend) CreatePool 

func (b *InMemoryBackend) CreatePool(in CreatePoolIn, host string) (poolWire, string, string, error)

CreatePool creates a stream pool; it is CREATING until the transition elapses.

func (*InMemoryBackend) CreateStream 

func (b *InMemoryBackend) CreateStream(in CreateStreamIn, endpoint string) (streamWire, string, string, error)

CreateStream creates a stream in the given pool, or in the compartment's default pool.

func (*InMemoryBackend) DeleteHarness 

func (b *InMemoryBackend) DeleteHarness(id, ifMatch string) (string, error)

DeleteHarness moves a connect harness to DELETING.

func (*InMemoryBackend) DeletePool 

func (b *InMemoryBackend) DeletePool(id, ifMatch string) (string, error)

DeletePool moves an empty pool to DELETING.

func (*InMemoryBackend) DeleteStream 

func (b *InMemoryBackend) DeleteStream(id, ifMatch string) (string, error)

DeleteStream moves a stream to DELETING; its messages are dropped when it becomes DELETED.

func (*InMemoryBackend) GetGroup 

func (b *InMemoryBackend) GetGroup(streamID, name string) (groupWire, error)

GetGroup returns the committed offsets and reservations of a consumer group.

func (*InMemoryBackend) GetHarness 

func (b *InMemoryBackend) GetHarness(id string) (harnessWire, string, error)

GetHarness returns one connect harness.

func (*InMemoryBackend) GetMessages 

func (b *InMemoryBackend) GetMessages(streamID, cursor string, limit int) ([]messageWire, string, error)

GetMessages reads from a plain or group cursor and returns the cursor for the next call.

func (*InMemoryBackend) GetPool 

func (b *InMemoryBackend) GetPool(id, host string) (poolWire, string, error)

GetPool returns one pool.

func (*InMemoryBackend) GetStream 

func (b *InMemoryBackend) GetStream(id, endpoint string) (streamWire, string, error)

GetStream returns one stream.

func (*InMemoryBackend) GetWorkRequest 

func (b *InMemoryBackend) GetWorkRequest(id string) (workRequestWire, error)

GetWorkRequest returns one work request.

func (*InMemoryBackend) ListHarnesses 

func (b *InMemoryBackend) ListHarnesses(
	compartment, id, name, state string,
	o ListOpts,
) (page.Page[harnessWire], error)

ListHarnesses lists the connect harnesses of a compartment.

func (*InMemoryBackend) ListPools 

func (b *InMemoryBackend) ListPools(
	compartment, id, name, state string,
	o ListOpts,
) (page.Page[poolSummaryWire], error)

ListPools lists the pools of a compartment.

func (*InMemoryBackend) ListStreams 

func (b *InMemoryBackend) ListStreams(
	compartment, poolID, id, name, state, endpoint string, o ListOpts,
) (page.Page[streamSummaryWire], error)

ListStreams lists streams by compartment, pool, id or name.

func (*InMemoryBackend) ListWorkRequests 

func (b *InMemoryBackend) ListWorkRequests(
	compartment, resourceID, id string, o ListOpts,
) (page.Page[workRequestWire], error)

ListWorkRequests lists the work requests of a compartment, optionally for one resource.

func (*InMemoryBackend) PartitionRanges 

func (b *InMemoryBackend) PartitionRanges(streamID string) ([]PartitionRange, error)

PartitionRanges reports every partition's offset window for a live stream.

func (*InMemoryBackend) PutMessages 

func (b *InMemoryBackend) PutMessages(streamID string, in putMessagesIn) (putResultWire, error)

PutMessages appends messages, choosing each partition by key hash (round-robin without a key).

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a snapshot.

func (*InMemoryBackend) SetClock 

func (b *InMemoryBackend) SetClock(now func() time.Time)

SetClock injects the time source used for lifecycle transitions, retention and reservations.

func (*InMemoryBackend) SetKeyChecker 

func (b *InMemoryBackend) SetKeyChecker(fn KeyChecker)

SetKeyChecker wires validation of customEncryptionKey references against a Vault backend.

func (*InMemoryBackend) SetTransition 

func (b *InMemoryBackend) SetTransition(d time.Duration)

SetTransition sets how long CREATING and DELETING states last.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state, including registered API keys.

func (*InMemoryBackend) Sweep 

func (b *InMemoryBackend) Sweep()

Sweep applies lifecycle transitions and retention against the backend clock.

func (*InMemoryBackend) UpdateGroup 

func (b *InMemoryBackend) UpdateGroup(streamID, name string, in UpdateGroupIn) error

UpdateGroup resets the committed offsets of every partition.

func (*InMemoryBackend) UpdateHarness 

func (b *InMemoryBackend) UpdateHarness(id, ifMatch string, tags Tags) (harnessWire, string, string, error)

UpdateHarness changes the tags of a connect harness.

func (*InMemoryBackend) UpdatePool 

func (b *InMemoryBackend) UpdatePool(id, ifMatch, host string, in UpdatePoolIn) (poolWire, string, string, error)

UpdatePool changes name, Kafka settings, encryption key and tags.

func (*InMemoryBackend) UpdateStream 

func (b *InMemoryBackend) UpdateStream(
	id, ifMatch, endpoint string, in UpdateStreamIn,
) (streamWire, string, string, error)

UpdateStream moves a stream to another pool and changes its tags.

func (*InMemoryBackend) WorkRequestExists 

func (b *InMemoryBackend) WorkRequestExists(id string) error

WorkRequestExists reports whether a work request exists (its error list is always empty).

func (*InMemoryBackend) WorkRequestLogs 

func (b *InMemoryBackend) WorkRequestLogs(id string, o ListOpts) (page.Page[wrLogWire], error)

WorkRequestLogs returns the log entries of a work request.

type KafkaSettings 

type KafkaSettings struct {
	BootstrapServers       *string `json:"bootstrapServers,omitempty"`
	AutoCreateTopicsEnable *bool   `json:"autoCreateTopicsEnable,omitempty"`
	LogRetentionHours      *int    `json:"logRetentionHours,omitempty"`
	NumPartitions          *int    `json:"numPartitions,omitempty"`
}

KafkaSettings are the Kafka-compatible settings of a stream pool.

type KeyChecker 

type KeyChecker func(keyID string) (state string, found bool)

KeyChecker reports the lifecycle state of a KMS key OCID and whether it exists.

type ListOpts 

type ListOpts struct {
	SortBy string
	Order  string
	Page   string
	Limit  int
}

ListOpts carries the shared list query parameters.

type Message 

type Message struct {
	Time  time.Time `json:"time"`
	Key   []byte    `json:"key,omitempty"`
	Value []byte    `json:"value"`
}

Message is one stored record; its offset is the partition base plus its index.

type Partition 

type Partition struct {
	Messages    []Message `json:"messages,omitempty"`
	Base        int64     `json:"base"`
	WindowSec   int64     `json:"windowSec"`
	WindowBytes int       `json:"windowBytes"`
}

Partition is one ordered log of a stream.

type PartitionRange 

type PartitionRange struct {
	Partition string `json:"partition"`
	Base      int64  `json:"base"`
	End       int64  `json:"end"`
}

PartitionRange is the readable offset window [Base, End) of one partition.

type Pool 

type Pool struct {
	TimeCreated   time.Time        `json:"timeCreated"`
	TransitionAt  *time.Time       `json:"transitionAt,omitempty"`
	Endpoint      *PrivateEndpoint `json:"privateEndpoint,omitempty"`
	Tags          Tags             `json:"tags"`
	Kafka         KafkaSettings    `json:"kafka"`
	ID            string           `json:"id"`
	CompartmentID string           `json:"compartmentId"`
	Name          string           `json:"name"`
	State         string           `json:"state"`
	KMSKeyID      string           `json:"kmsKeyId,omitempty"`
	Etag          string           `json:"etag"`
	Seq           int64            `json:"seq"`
	Private       bool             `json:"private,omitempty"`
	IsDefault     bool             `json:"isDefault,omitempty"`
}

Pool is a stream pool.

type PrivateEndpoint 

type PrivateEndpoint struct {
	SubnetID          *string  `json:"subnetId,omitempty"`
	PrivateEndpointIP *string  `json:"privateEndpointIp,omitempty"`
	NsgIDs            []string `json:"nsgIds,omitempty"`
}

PrivateEndpoint holds the private endpoint settings of a stream pool.

type Provider 

type Provider struct{}

Provider implements service.Provider for OCI Streaming. It runs on its own listener (see Handler.StartWorker) and never matches on the shared AWS router.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init creates the backend and handler; the port is bound in StartWorker.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type Reservation 

type Reservation struct {
	Until    time.Time `json:"until"`
	Instance string    `json:"instance"`
}

Reservation is one partition's consumer-group reservation.

type Settings 

type Settings struct {
	Port int `` //nolint:lll // config struct tags are intentionally verbose
	/* 214-byte string literal not displayed */
	EnforceAuth bool `` //nolint:lll // config struct tags are intentionally verbose
	/* 225-byte string literal not displayed */
}

Settings holds service-level configuration, embedded in the root CLI as CLI.OCIStreaming.

func DefaultSettings 

func DefaultSettings() Settings

DefaultSettings returns the settings used when no ConfigProvider is available.

type Stream 

type Stream struct {
	TimeCreated   time.Time    `json:"timeCreated"`
	TransitionAt  *time.Time   `json:"transitionAt,omitempty"`
	Tags          Tags         `json:"tags"`
	ID            string       `json:"id"`
	PoolID        string       `json:"poolId"`
	CompartmentID string       `json:"compartmentId"`
	Name          string       `json:"name"`
	State         string       `json:"state"`
	Etag          string       `json:"etag"`
	Parts         []*Partition `json:"partitions"`
	Retention     int          `json:"retentionInHours"`
	Seq           int64        `json:"seq"`
	RoundRobin    int          `json:"roundRobin"`
}

Stream is a partitioned message log.

type Tags 

type Tags struct {
	FreeformTags       map[string]string         `json:"freeformTags,omitempty"`
	DefinedTags        map[string]map[string]any `json:"definedTags,omitempty"`
	SecurityAttributes map[string]map[string]any `json:"securityAttributes,omitempty"`
}

Tags are the shared freeform and defined tag sets of a resource.

type UpdateGroupIn 

type UpdateGroupIn struct {
	Time *time.Time `json:"time"`
	Type string     `json:"type"`
}

UpdateGroupIn is the UpdateGroup request body.

type UpdatePoolIn 

type UpdatePoolIn struct {
	Name  *string        `json:"name"`
	Kafka *KafkaSettings `json:"kafkaSettings"`
	Key   *kmsKeyIn      `json:"customEncryptionKeyDetails"`
	Tags
}

UpdatePoolIn is the UpdateStreamPool request body.

type UpdateStreamIn 

type UpdateStreamIn struct {
	PoolID *string `json:"streamPoolId"`
	Tags
}

UpdateStreamIn is the UpdateStream request body.

type WorkRequest 

type WorkRequest struct {
	Accepted      time.Time `json:"accepted"`
	FinishAt      time.Time `json:"finishAt"`
	ID            string    `json:"id"`
	Operation     string    `json:"operation"`
	CompartmentID string    `json:"compartmentId"`
	EntityType    string    `json:"entityType"`
	EntityID      string    `json:"entityId"`
	EntityURI     string    `json:"entityUri"`
	Action        string    `json:"action"`
	Seq           int64     `json:"seq"`
}

WorkRequest records an asynchronous admin operation.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.