### Example: Create ProtoObservationCodec

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Instantiate a new observation codec with zstd compression enabled. Handle potential errors during setup.

```go
codec, err := llo.NewProtoObservationCodec(logger, true)
if err != nil {
    log.Fatalf("Failed to create codec: %v", err)
}
```

--------------------------------

### Create and Start RPC Client

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Demonstrates how to create a new gRPC client using `NewClient` and start its service. Ensure all necessary dependencies like logger and keys are provided.

```go
import (
    "crypto/ed25519"
    "github.com/smartcontractkit/chainlink-data-streams/rpc"
)

privateKey := ed25519.PrivateKey(privateKeyBytes)
serverPubKey := ed25519.PublicKey(serverPubKeyBytes)

client := rpc.NewClient(rpc.ClientOpts{
    Logger: logger,
    ClientSigner: privateKey,
    ServerPubKey: serverPubKey,
    ServerURL: "mercury.example.com:7080",
})

// Start the client service
if err := client.Start(ctx); err != nil {
    log.Fatalf("Failed to start client: %v", err)
}
```

--------------------------------

### Example DataSource Implementation

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/data-source.md

An example implementation of the DataSource interface using HTTP to fetch price data. It iterates through streams, makes GET requests, decodes the response, and updates stream values.

```go
type MyDataSource struct {
    client *http.Client
    urls   map[uint32]string
}

func (ds *MyDataSource) Observe(
    ctx context.Context,
    streamValues llo.StreamValues,
    opts llo.DSOpts,
) error {
    for streamID := range streamValues {
        url, ok := ds.urls[streamID]
        if !ok {
            continue
        }
        
        req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
        resp, err := ds.client.Do(req)
        if err != nil {
            // Log but continue with other streams
            continue
        }
        
        var price decimal.Decimal
        if err := json.NewDecoder(resp.Body).Decode(&price); err != nil {
            continue
        }
        
        streamValues[streamID] = llo.ToDecimal(price)
    }
    
    return nil
}
```

--------------------------------

### Configuration Validation Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Demonstrates how to validate a PluginConfig and iterate through validation errors.

```go
cfg := llo.config.PluginConfig{...}
if err := cfg.Validate(); err != nil {
    for _, validationErr := range errors.Unwrap(err).(interface{ Unwrap() []error }).Unwrap() {
        log.Printf("Validation error: %v", validationErr)
    }
}
```

--------------------------------

### EVMOnchainConfigCodec Encode Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Encodes on-chain configuration to the 64-byte EVM format. This example shows how to create an OnchainConfig and encode it.

```go
config := llo.OnchainConfig{
    Version: 1,
    PredecessorConfigDigest: &previousDigest,
}

encoded, err := codec.Encode(config)
if err != nil {
    log.Fatalf("Encoding failed: %v", err)
}
```

--------------------------------

### Server Address Format Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Illustrates the required format for server URLs, which is host:port.

```plaintext
host:port
```

--------------------------------

### Initialize RPC Client

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Example of how to initialize a new RPC client using the ClientOpts. This requires valid private and public keys, a logger, and the server's URL.

```go
clientPrivateKey := ed25519.PrivateKey(keyBytes)
serverPublicKey := ed25519.PublicKey(pubKeyBytes)

client := rpc.NewClient(rpc.ClientOpts{
    Logger: logger,
    ClientSigner: clientPrivateKey,
    ServerPubKey: serverPublicKey,
    ServerURL: "mercury.example.com:7080",
})
```

--------------------------------

### PluginConfig Validation Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Demonstrates how to instantiate and validate a PluginConfig in Go. Ensure all required fields are set and validation rules are met before use.

```go
cfg := llo.config.PluginConfig{
    ChannelDefinitionsContractAddress: common.HexToAddress("0x123..."),
    ChannelDefinitionsContractFromBlock: 18000000,
    BenchmarkMode: false,
    KeyBundleIDs: map[string]string{
        "evm": "bundle-1",
    },
    DonID: 1,
    Servers: map[string]hex.PlainHexBytes{
        "wss://mercury.example.com:4242": serverPubKeyBytes,
    },
}

if err := cfg.Validate(); err != nil {
    log.Fatalf("Invalid config: %v", err)
}
```

--------------------------------

### Example Multiplier Options JSON

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Provides an example JSON structure for specifying multipliers, including stream IDs and their corresponding scaling factors.

```json
{
  "multipliers": [
    {
      "streamID": 1000000001,
      "multiplier": "10000"
    },
    {
      "streamID": 1000000002,
      "multiplier": "1000000"
    }
  ]
}
```

--------------------------------

### ModeAggregator Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/aggregators.md

Demonstrates how to use ModeAggregator with a slice of StreamValue and a maximum Byzantine failures integer.

```go
observations := []llo.StreamValue{
    llo.ToDecimal(decimal.NewFromString("100.00")),
    llo.ToDecimal(decimal.NewFromString("100.00")),
    llo.ToDecimal(decimal.NewFromString("99.00")),
}

f := 1
mode, err := llo.ModeAggregator(observations, f)
// mode = Decimal(100.00) - appears twice, meets f+1 threshold
```

--------------------------------

### ReportCodecCapabilityTrigger Verify Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Shows how to instantiate the codec and use the Verify method to check channel definition compatibility. Logs a fatal error if the definition is invalid.

```go
codec := cre.NewReportCodecCapabilityTrigger(logger, donID)

err := codec.Verify(channelDefinition)
if err != nil {
    log.Fatalf("Channel definition invalid: %v", err)
}
```

--------------------------------

### Start

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Initializes the gRPC connection with mTLS credentials. It establishes secure communication channels and configures aggressive reconnect parameters for reliability.

```APIDOC
## Start

### Description
Initializes the gRPC connection with mTLS credentials. It establishes secure communication channels and configures aggressive reconnect parameters for reliability.

### Method
`Start`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
#### Context Parameter
- **ctx** (context.Context) - Required - Context for initialization timeout

### Return Type
Returns error if connection fails or TLS setup fails.

### Behavior
- Establishes mTLS credentials using client and server keys
- Creates gRPC connection with aggressive reconnect parameters:
  - Base delay: 1 second
  - Exponential backoff with 2x multiplier
  - Max delay: 30 seconds
  - Jitter: 20%
- Minimum connect timeout: 1 second
- Keepalive parameters:
  - Time: 10 seconds
  - Timeout: 20 seconds
  - PermitWithoutStream: true
```

--------------------------------

### Example: Decode Observation

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Decode binary data into an Observation struct. Ensure proper error handling for potential decoding issues.

```go
observation, err := codec.Decode(binaryData)
if err != nil {
    log.Errorf("Decoding failed: %v", err)
}
```

--------------------------------

### Decode Outcome Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Demonstrates how to decode binary outcome data back into an Outcome object using an OutcomeCodec.

```go
outcome, err := outcomeCodec.Decode(binaryData)
if err != nil {
    log.Errorf("Decoding failed: %v", err)
}
```

--------------------------------

### Go Client Example for Transmit

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Demonstrates how to use the Go client to transmit a signed report to the Mercury server. Ensure you have the necessary logger, private key, server public key, and server URL configured.

```go
import "github.com/smartcontractkit/chainlink-data-streams/rpc"

client := rpc.NewClient(rpc.ClientOpts{
    Logger: logger,
    ClientSigner: privateKey,
    ServerPubKey: serverPubKey,
    ServerURL: "mercury.example.com:7080",
})

request := &rpc.TransmitRequest{
    Payload: reportBytes,
    ReportFormat: 1, // format ID
}

response, err := client.Transmit(ctx, request)
if err != nil {
    log.Errorf("Transmit failed: %v", err)
} else if response.Code != 0 {
    log.Warnf("Server error: %s", response.Error)
} else {
    log.Info("Report transmitted")
}
```

--------------------------------

### NewTransmitter Constructor Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/cre-transmitter.md

Demonstrates how to create a new CRE transmitter instance using the TransmitterConfig. Ensure all required fields like Logger, CapabilitiesRegistry, and DonID are provided.

```go
transmitter, err := cre.TransmitterConfig{
    Logger: logger,
    CapabilitiesRegistry: capRegistry,
    DonID: 1,
    TriggerCapabilityName: "streams-trigger",
    TriggerCapabilityVersion: "2.0.0",
    TriggerTickerMinResolutionMs: 1000,
}.NewTransmitter()
if err != nil {
    log.Fatalf("Failed to create transmitter: %v", err)
}
```

--------------------------------

### Start RPC Client Service

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Initializes the gRPC connection with mTLS credentials. Establishes credentials, creates a gRPC connection with aggressive reconnect parameters, and sets keepalive parameters.

```go
func (c *client) Start(ctx context.Context) error
```

--------------------------------

### ReportCodecCapabilityTrigger Encode Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Demonstrates how to use the Encode method with specified multiplier options. Handles potential encoding errors.

```go
opts := cre.ReportCodecCapabilityTriggerOpts{
    Multipliers: []cre.ReportCodecCapabilityTriggerMultiplier{
        {
            StreamID: 1000000001,
            Multiplier: decimal.NewFromInt(10000),
        },
    },
}

codec := cre.NewReportCodecCapabilityTrigger(logger, donID)

encoded, err := codec.Encode(report, channelDef, optsCache)
if err != nil {
    log.Errorf("Encoding failed: %v", err)
}
```

--------------------------------

### MedianAggregator Example Usage

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/aggregators.md

Demonstrates how to use the MedianAggregator function with a slice of Decimal StreamValues to find the median, allowing for one Byzantine fault.

```go
observedValues := []llo.StreamValue{
    llo.ToDecimal(decimal.NewFromString("99.50")),
    llo.ToDecimal(decimal.NewFromString("100.00")),
    llo.ToDecimal(decimal.NewFromString("100.50")),
}

f := 1 // Allow 1 Byzantine node (n=3 total)
aggregated, err := llo.MedianAggregator(observedValues, f)
if err != nil {
    log.Errorf("Median aggregation failed: %v", err)
}
// aggregated = Decimal(100.00)
```

--------------------------------

### Get Configured Server URL

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Retrieves the server URL that the client is configured to communicate with. This is useful for logging or debugging.

```go
func (c *client) ServerURL() string
```

```go
url := client.ServerURL()
fmt.Printf("Transmitting to: %s\n", url)
```

--------------------------------

### Example: Encode Observation

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Encode a sample observation containing a timestamp and stream values. Check for errors during the encoding process.

```go
observation := llo.Observation{
    UnixTimestampNanoseconds: uint64(time.Now().UnixNano()),
    StreamValues: llo.StreamValues{
        1000000001: llo.ToDecimal(decimal.NewFromString("99.50")),
    },
}

encoded, err := codec.Encode(observation)
if err != nil {
    log.Errorf("Encoding failed: %v", err)
}
```

--------------------------------

### cURL Example for Transmit

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Shows how to call the Transmit RPC method using grpcurl, assuming HTTP/2 support. Replace placeholders with actual base64-encoded payload and report format.

```bash
grpcurl -plaintext \
  -d '{"payload": "<base64-encoded-bytes>", "reportFormat": 1}' \
  mercury.example.com:7080 rpc.Transmitter/Transmit
```

--------------------------------

### Encode Outcome Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Demonstrates how to encode an Outcome object into protobuf binary format using an OutcomeCodec. Ensure outcomes encode deterministically for consensus.

```go
outcome := llo.Outcome{
    LifeCycleStage: "production",
    ObservationTimestampNanoseconds: uint64(time.Now().UnixNano()),
    ChannelDefinitions: channelDefs,
    StreamAggregates: streamAggregates,
    ValidAfterNanoseconds: validAfterMap,
}

encoded, err := outcomeCodec.Encode(outcome)
if err != nil {
    log.Errorf("Encoding failed: %v", err)
}
```

--------------------------------

### Implement Backoff and Retry Logic

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Manually implement retry logic with exponential backoff for failed transmissions, as the client does not automatically retry. This example attempts up to 3 retries with increasing delays.

```go
for attempt := 0; attempt < 3; attempt++ {
    resp, err := client.Transmit(ctx, request)
    if err == nil {
        break
    }
    time.Sleep(time.Duration(math.Pow(2, float64(attempt))) * time.Second)
}
```

--------------------------------

### Channel Definition Example

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Defines how streams are mapped to reports, specifying the report format, stream IDs, aggregator types, and optional settings. Use this to configure data stream reporting.

```json
{
  "reportFormat": "capability_trigger",
  "streams": [
    {"streamID": 1000000001, "aggregator": 1},
    {"streamID": 1000000002, "aggregator": 2}
  ],
  "opts": {...format-specific options...},
  "tombstone": false
}
```

--------------------------------

### EVMOnchainConfigCodec.Encode

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Encodes on-chain configuration into a 64-byte EVM format. An example demonstrates encoding a configuration with version 1 and a predecessor digest.

```APIDOC
## EVMOnchainConfigCodec.Encode

### Description
Encodes on-chain configuration to 64-byte EVM format.

### Example
```go
config := llo.OnchainConfig{
    Version: 1,
    PredecessorConfigDigest: &previousDigest,
}

encoded, err := codec.Encode(config)
if err != nil {
    log.Fatalf("Encoding failed: %v", err)
}
```
```

--------------------------------

### Transmit Reports via gRPC

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Transmit reports using gRPC by creating a client, starting it, and then calling the Transmit method with the report payload and format. Ensure the client is closed after use.

```go
import (
	"context"

	"github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink/rpc"
)

// Create client
client := rpc.NewClient(rpc.ClientOpts{
    Logger: logger,
    ClientSigner: privateKey,
    ServerPubKey: serverPubKey,
    ServerURL: "mercury.example.com:7080",
})

// Start client
if err := client.Start(ctx); err != nil {
    log.Fatalf("Start failed: %v", err)
}
defer client.Close()

// Transmit report
response, err := client.Transmit(ctx, &rpc.TransmitRequest{
    Payload: reportBytes,
    ReportFormat: 1,
})
if err != nil {
    log.Errorf("Transmit failed: %v", err)
}
```

--------------------------------

### Example Channel Definition JSON

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/channel-definitions.md

This JSON object defines a channel with two streams, specifying their IDs and aggregator modes. It also includes options for multipliers for each stream.

```json
{
  "reportFormat": "capability_trigger",
  "streams": [
    {
      "streamId": 1000000001,  // Mode aggregator
      "aggregator": 1
    },
    {
      "streamId": 1000000002,
      "aggregator": 1
    }
  ],
  "opts": {
    "multipliers": [
      {
        "streamId": 1000000001,
        "multiplier": "10000"
      },
      {
        "streamId": 1000000002,
        "multiplier": "1000000"
      }
    ]
  },
  "tombstone": false,
  "source": 0,
  "disableNilStreamValues": false
}
```

--------------------------------

### Configure Plugin with Channel Definitions and Servers

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Configure the plugin with essential details like the channel definitions contract address, benchmark mode, key bundle IDs, and server public keys. Always validate the configuration before use.

```go
import (
	"encoding/hex"

	"github.com/ethereum/go-ethereum/common"
	"github.com/smartcontractkit/chainlink-common/pkg/types/config"
)

pluginConfig := config.PluginConfig{
    ChannelDefinitionsContractAddress: common.HexToAddress("0x..."),
    BenchmarkMode: false,
    KeyBundleIDs: map[string]string{
        "evm": "bundle-id-1",
    },
    DonID: 1,
    Servers: map[string]hex.PlainHexBytes{
        "wss://mercury.example.com:4242": serverPubKeyBytes,
    },
}

if err := pluginConfig.Validate(); err != nil {
    log.Fatalf("Invalid config: %v", err)
}
```

--------------------------------

### Instantiate Reporting Plugin

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Create an instance of the reporting plugin using the factory. This step requires the context and configuration for the plugin.

```go
// 3. Create plugin instance from factory
plugin, pluginInfo, err := factory.NewReportingPlugin(ctx, cfg)
```

--------------------------------

### Create a New Reporting Plugin

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/plugin-factory.md

Instantiates a new reporting plugin using the provided context and configuration. Ensure all required configuration fields are correctly populated.

```go
func (f *PluginFactory) NewReportingPlugin(
    ctx context.Context,
    cfg ocr3types.ReportingPluginConfig,
) (ocr3types.ReportingPlugin[llotypes.ReportInfo], ocr3types.ReportingPluginInfo, error)
```

```go
reportingPlugin, pluginInfo, err := factory.NewReportingPlugin(ctx, ocr3types.ReportingPluginConfig{
    ConfigDigest: configDigest,
    N: 10,
    F: 3,
    OnchainConfig: onchainConfigBytes,
    OffchainConfig: offchainConfigBytes,
    MaxDurationObservation: 2 * time.Second,
})
if err != nil {
    log.Fatalf("failed to create reporting plugin: %v", err)
}
```

--------------------------------

### Create Reporting Plugin Factory

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Create a plugin factory with necessary dependencies like configuration, logger, data source, and codecs. This factory is used to instantiate the reporting plugin.

```go
import (
	"context"

	llotypes "github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink"
	"github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink/codecs"
	"github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink/codecs/evm"
	"github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink/llo"
)

// 2. Create factory with dependencies
factory := llo.NewPluginFactory(llo.PluginFactoryParams{
    Config: llo.Config{VerboseLogging: false},
    Logger: logger,
    DataSource: myDataSource,
    ChannelDefinitionCache: channelCache,
    OnchainConfigCodec: llo.EVMOnchainConfigCodec{},
    ReportCodecs: map[llotypes.ReportFormat]llo.ReportCodec{
        llotypes.ReportFormatCapabilityTrigger: reportCodec,
    },
    DonID: 1,
})
```

--------------------------------

### Build and Run Mercury Integration Tests

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/mercury/cmd/chainlink-mercury/README.md

Builds the mercury binary and then runs integration tests, redirecting output to a log file. Ensure CL_MERCURY_CMD is set to the fully resolved binary path.

```shell
go install # builds `mercury` binary in this dir
CL_MERCURY_CMD=chainlink-mercury go test -v -timeout 120s -run ^TestIntegration_MercuryV github.com/smartcontractkit/chainlink/v2/core/services/ocr2/plugins/mercury 2>&1 | tee /tmp/mercury_loop.log
```

--------------------------------

### Get Aggregator Function

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/aggregators.md

Retrieves the appropriate aggregator function based on the provided llotypes.Aggregator enum. Returns nil if the aggregator type is unknown.

```go
func GetAggregatorFunc(a llotypes.Aggregator) AggregatorFunc
```

--------------------------------

### Configure Observation Compression

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Enable observation compression in OffchainConfig for networks with limited bandwidth. This can reduce data transfer costs.

```go
offchainCfg := llo.OffchainConfig{
    ProtocolVersion: 1,
    DefaultMinReportIntervalNanoseconds: 1_000_000_000, // 1 second
    EnableObservationCompression: true,
}
```

--------------------------------

### RPC Client Configuration Options

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Defines the structure for configuring an RPC client. Ensure all fields, including logger, signer, server public key, and server URL, are correctly provided.

```go
type ClientOpts struct {
    Logger logger.Logger
    ClientSigner crypto.Signer
    ServerPubKey ed25519.PublicKey
    ServerURL string
}
```

--------------------------------

### NewClient Constructor

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Creates a new gRPC client for transmitting reports. It requires a ClientOpts struct containing logger, signer, server public key, and server URL.

```APIDOC
## NewClient

```go
func NewClient(opts ClientOpts) Client
```

Creates a new gRPC client for transmitting reports.

### Parameters

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| opts | ClientOpts | Yes | Client configuration options |

### Return Type

Returns `Client` interface implementation.

### ClientOpts Structure

```go
type ClientOpts struct {
    Logger logger.Logger
    ClientSigner crypto.Signer
    ServerPubKey ed25519.PublicKey
    ServerURL string
}
```

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| Logger | logger.Logger | Yes | Logger instance for diagnostics |
| ClientSigner | crypto.Signer | Yes | Ed25519 private key for signing requests (implements crypto.Signer) |
| ServerPubKey | ed25519.PublicKey | Yes | Ed25519 public key of the Mercury server |
| ServerURL | string | Yes | Server address (format: "host:port") |

### Example

```go
import (
    "crypto/ed25519"
    "github.com/smartcontractkit/chainlink-data-streams/rpc"
)

privateKey := ed25519.PrivateKey(privateKeyBytes)
serverPubKey := ed25519.PublicKey(serverPubKeyBytes)

client := rpc.NewClient(rpc.ClientOpts{
    Logger: logger,
    ClientSigner: privateKey,
    ServerPubKey: serverPubKey,
    ServerURL: "mercury.example.com:7080",
})

// Start the client service
if err := client.Start(ctx); err != nil {
    log.Fatalf("Failed to start client: %v", err)
}
```
```

--------------------------------

### NewPluginFactory Constructor

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/plugin-factory.md

Creates a new PluginFactory instance. Requires PluginFactoryParams, which includes configuration, logger, data source, and other essential components.

```go
func NewPluginFactory(p PluginFactoryParams) *PluginFactory
```

```go
factory := llo.NewPluginFactory(llo.PluginFactoryParams{
    Config: llo.Config{
        VerboseLogging: false,
    },
    Logger: logger,
    DataSource: dataSource,
    ChannelDefinitionCache: channelCache,
    OnchainConfigCodec: llo.EVMOnchainConfigCodec{},
    ReportCodecs: map[llotypes.ReportFormat]llo.ReportCodec{
        llotypes.ReportFormatEVMABIEncoded: reportCodec,
    },
    DonID: 1,
})
```

--------------------------------

### EVMOnchainConfigCodec Decode Method

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Decodes on-chain configuration from a 64-byte EVM format. The format includes a version and a predecessor configuration digest. The input must be exactly 64 bytes, the version must be 1, and it must fit within a uint8.

```go
func (EVMOnchainConfigCodec) Decode(b []byte) (OnchainConfig, error)
```

--------------------------------

### Observe Stream Values

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Use this pattern to observe stream values from a data source. Ensure proper context and options are provided.

```go
opts := &mockDSOpts{}
streamValues := make(llo.StreamValues)
err := dataSource.Observe(ctx, streamValues, opts)
if err != nil {
    t.Fatalf("Observe failed: %v", err)
}
```

--------------------------------

### Configure Multiple Report Codecs

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Configures the plugin with multiple codecs for different report formats, such as CapabilityTrigger and EVM ABI encoded. The appropriate codec is selected based on the channel definition's ReportFormat during report generation.

```go
ReportCodecs: map[llotypes.ReportFormat]ReportCodec{
    llotypes.ReportFormatCapabilityTrigger: cre.NewReportCodecCapabilityTrigger(logger, donID),
    llotypes.ReportFormatEVMABIEncoded: evmCodec,
    // ... other formats
}
```

--------------------------------

### Set Request Context Timeout

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Implement request timeouts by creating a context with a specified duration using `context.WithTimeout`. This prevents requests from hanging indefinitely and helps manage network latency.

```go
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
response, err := client.Transmit(ctx, request)
```

--------------------------------

### Implement DataSource Interface for Reporting Plugin

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Implement the DataSource interface to define how your plugin fetches and populates stream values. This is the first step in creating a reporting plugin.

```go
import (
	"context"

	llo "github.com/smartcontractkit/chainlink-common/pkg/types/liquid-chainlink"
)

// 1. Implement DataSource interface
type MyDataSource struct {}
func (ds *MyDataSource) Observe(ctx context.Context, streamValues llo.StreamValues, opts llo.DSOpts) error {
    // Fetch and populate stream values
    return nil
}
```

--------------------------------

### PluginConfig JSON/TOML Format

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Illustrates the expected JSON or TOML format for the PluginConfig, detailing each field and its expected values.

```json
{
  "channelDefinitionsContractAddress": "0x1234567890123456789012345678901234567890",
  "channelDefinitionsContractFromBlock": 18000000,
  "channelDefinitions": "{...json channel definitions...}",
  "benchmarkMode": false,
  "keyBundleIDs": {
    "evm": "bundle-id-1",
    "solana": "bundle-id-2"
  },
  "donID": 1,
  "servers": {
    "wss://mercury-1.example.com:4242": "a1b2c3d4...",
    "wss://mercury-2.example.com:4242": "e5f6g7h8..."
  },
  "transmitters": [
    {
      "type": "cre",
      "opts": {...}
    }
  ]
}
```

--------------------------------

### Observe Method Signature

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/data-source.md

Fetches observations for all known streams and populates the streamValues map. Returns an error if observation fails.

```go
func Observe(
    ctx context.Context,
    streamValues StreamValues,
    opts DSOpts,
) error
```

--------------------------------

### Onchain Config Codec Interface

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Interface for serializing and deserializing onchain configuration data.

```go
type OnchainConfigCodec interface {
    Decode(b []byte) (OnchainConfig, error)
    Encode(c OnchainConfig) ([]byte, error)
}
```

--------------------------------

### NewReportingPlugin

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/plugin-factory.md

Creates a new ReportingPlugin instance for the given OCR3 configuration. It takes a context and a reporting plugin configuration, returning the initialized plugin, its metadata, and any potential errors.

```APIDOC
## NewReportingPlugin

### Description
Creates a new ReportingPlugin instance for the given OCR3 configuration.

### Method
`NewReportingPlugin`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
reportingPlugin, pluginInfo, err := factory.NewReportingPlugin(ctx, ocr3types.ReportingPluginConfig{
    ConfigDigest: configDigest,
    N: 10,
    F: 3,
    OnchainConfig: onchainConfigBytes,
    OffchainConfig: offchainConfigBytes,
    MaxDurationObservation: 2 * time.Second,
})
if err != nil {
    log.Fatalf("failed to create reporting plugin: %v", err)
}
```

### Response
#### Success Response
- `ocr3types.ReportingPlugin[llotypes.ReportInfo]` — The initialized reporting plugin
- `ocr3types.ReportingPluginInfo` — Plugin metadata including protocol limits

#### Response Example
None provided in source.

ERROR HANDLING:
- Onchain config decode fails
- Offchain config decode fails
- Observation codec creation fails
```

--------------------------------

### LatestReport Go Client Implementation (Not Supported)

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

This Go code snippet shows the current implementation for the LatestReport method in the gRPC client, which indicates that the feature is not supported.

```go
func (c *client) LatestReport(ctx context.Context, req *LatestReportRequest) (resp *LatestReportResponse, err error) {
    return nil, errors.New("LatestReport is not supported in grpc mode")
}
```

--------------------------------

### PluginFactory Configuration Structure

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/plugin-factory.md

Defines the configuration structure for the PluginFactory. VerboseLogging enables detailed logging for debugging purposes.

```go
type Config struct {
    VerboseLogging bool
}
```

--------------------------------

### OnchainConfigCodec Interface

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Defines the interface for encoding and decoding on-chain configurations.

```go
type OnchainConfigCodec interface {
    Decode(b []byte) (OnchainConfig, error)
    Encode(c OnchainConfig) ([]byte, error)
}

type EVMOnchainConfigCodec struct {}
```

--------------------------------

### PluginConfig Structure

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Defines the configuration structure for the LLO plugin, including contract addresses, benchmark mode, key bundles, servers, and transmitters.

```go
type PluginConfig struct {
    ChannelDefinitionsContractAddress common.Address
    ChannelDefinitionsContractFromBlock int64
    ChannelDefinitions string
    BenchmarkMode bool
    KeyBundleIDs map[string]string
    DonID uint32
    Servers map[string]hex.PlainHexBytes
    Transmitters []TransmitterConfig
}
```

--------------------------------

### NewReportCodecCapabilityTrigger Constructor

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Creates a new capability trigger report codec. Requires a logger instance and a DON identifier for error context.

```go
func NewReportCodecCapabilityTrigger(lggr logger.Logger, donID uint32) ReportCodecCapabilityTrigger
```

--------------------------------

### ProtoOutcomeCodecV1 Implementation

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Current implementation of OutcomeCodec using LLOOutcomeProtoV1. It supports nanosecond precision for ValidAfterNanoseconds and uint64 for observation timestamps.

```go
type protoOutcomeCodecV1 struct {}
```

--------------------------------

### Validate Channel Definitions

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

This pattern validates channel definitions using provided codecs. Ensure the channel definition and codecs are correctly set up.

```go
defs := llotypes.ChannelDefinitions{
    1: channelDef,
}
err := llo.VerifyChannelDefinitions(codecs, defs)
if err != nil {
    t.Errorf("Channel definition invalid: %v", err)
}
```

--------------------------------

### ProtoOutcomeCodecV0 Implementation

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Legacy implementation of OutcomeCodec using LLOOutcomeProtoV0. It uses ValidAfterSeconds and int64 for observation timestamps.

```go
type protoOutcomeCodecV0 struct {}
```

--------------------------------

### Protocol Constants

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

These constants represent hard constraints enforced by the plugin for various data sizes and counts within the protocol.

```go
const (
    MaxReportCount = 5
    MaxObservationLength = 32 * 1024
    MaxOutcomeLength = 256 * 1024
    MaxReportLength = 128 * 1024
    MaxObservationRemoveChannelIDsLength = 5
    MaxObservationUpdateChannelDefinitionsLength = 5
    MaxObservationStreamValuesLength = 10_000
    MaxStreamsPerChannel = 10_000
    MaxOutcomeChannelDefinitionsLength = 5
)
```

--------------------------------

### LLOTriggerConfig Structure

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/cre-transmitter.md

Defines the configuration for LLO trigger registration, including a list of stream IDs and the maximum frequency in milliseconds.

```go
type LLOTriggerConfig struct {
    StreamIDs []LLOStreamID
    MaxFrequencyMs uint64
}
```

--------------------------------

### NewProtoObservationCodec Function

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Creates a new observation codec using protobuf serialization. Enable compression for potentially smaller payloads.

```go
type protoObservationCodec struct {
    logger logger.Logger
    enableCompression bool
    compressor *compressor
}

func NewProtoObservationCodec(
    lggr logger.Logger,
    enableCompression bool,
) (ObservationCodec, error)
```

--------------------------------

### OnchainConfig Struct

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Represents configuration data stored directly on the blockchain. It includes the version and a reference to the predecessor's configuration digest.

```go
type OnchainConfig struct {
    Version uint8
    PredecessorConfigDigest *types.ConfigDigest
}
```

--------------------------------

### LLOutcomeProtoV0 Message Format

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Defines the protobuf message structure for LLO outcome version 0, including lifecycle stage, observation timestamp, and channel definitions.

```protobuf
message LLOOutcomeProtoV0 {
    string lifeCycleStage = 1;
    int64 observationTimestampNanoseconds = 2;
    repeated LLOChannelIDAndDefinitionProto channelDefinitions = 3;
    repeated LLOChannelIDAndValidAfterSecondsProto validAfterSeconds = 4;
    repeated LLOStreamAggregate streamAggregates = 5;
}
```

--------------------------------

### Encode and Decode Data

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Use this pattern for encoding observations into a byte slice and decoding them back. It utilizes a ProtoObservationCodec.

```go
codec, _ := llo.NewProtoObservationCodec(logger, false)
encoded, _ := codec.Encode(observation)
decoded, err := codec.Decode(encoded)
if err != nil {
    t.Fatalf("Decode failed: %v", err)
}
```

--------------------------------

### ServerURL

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/rpc-client.md

Returns the configured server URL for the RPC client.

```APIDOC
## ServerURL

### Description
Returns the configured server URL.

### Method
*Not applicable (gRPC method)*

### Endpoint
*Not applicable (gRPC method)*

### Parameters
*None*

### Request Example
```go
url := client.ServerURL()
fmt.Printf("Transmitting to: %s\n", url)
```

### Response
#### Success Response
- Returns the server address as a string.

#### Response Example
```
"http://example.com/rpc"
```
```

--------------------------------

### OffchainConfig Struct

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Represents configuration data stored off-chain, which is decoded by the plugin. It specifies protocol version, minimum report interval, and observation compression settings.

```go
type OffchainConfig struct {
    ProtocolVersion uint32
    DefaultMinReportIntervalNanoseconds uint64
    EnableObservationCompression bool
}
```

--------------------------------

### LLOutcomeProtoV1 Message Format

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Defines the protobuf message structure for LLO outcome version 1, including lifecycle stage, observation timestamp, and channel definitions with nanosecond precision.

```protobuf
message LLOOutcomeProtoV1 {
    string lifeCycleStage = 1;
    uint64 observationTimestampNanoseconds = 2;
    repeated LLOChannelIDAndDefinitionProto channelDefinitions = 3;
    repeated LLOChannelIDAndValidAfterNanosecondsProto validAfterNanoseconds = 4;
    repeated LLOStreamAggregate streamAggregates = 5;
}
```

--------------------------------

### PluginConfig Structure

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/configuration.md

Defines the structure for Mercury server configuration, including server URLs, public keys, and feed identifiers.

```go
type PluginConfig struct {
    RawServerURL string
    ServerPubKey hex.PlainHexBytes
    Servers map[string]hex.PlainHexBytes
    InitialBlockNumber null.Int64
    LinkFeedID *mercury.FeedID
    NativeFeedID *mercury.FeedID
}

type Server struct {
    URL string
    PubKey hex.PlainHexBytes
}
```

--------------------------------

### Create Query for Leader to Followers

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/reporting-plugin.md

Creates a Query to be sent from the leader to follower nodes during an observation request. Currently returns a nil query as oracles do not need to coordinate observations. The context may expire, so any blocking operations should stop immediately.

```go
query, err := plugin.Query(ctx, outcomeContext)
if err != nil {
    return err
}
// query is nil for LLO plugin
```

--------------------------------

### Check Plugin Creation Error

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Always check the error returned when creating a new reporting plugin. This ensures the plugin is initialized correctly before use.

```go
plugin, _, err := factory.NewReportingPlugin(ctx, cfg)
if err != nil {
    log.Errorf("Plugin creation failed: %v", err)
    return err
}
```

--------------------------------

### OffchainConfig GetOutcomeCodec Method

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Selects the appropriate OutcomeCodec based on the OffchainConfig's ProtocolVersion. Defaults to V1 for versions other than 0.

```go
func (c OffchainConfig) GetOutcomeCodec() OutcomeCodec {
    switch c.ProtocolVersion {
    case 0:
        return protoOutcomeCodecV0{}
    default:
        return protoOutcomeCodecV1{}
    }
}
```

--------------------------------

### Lifecycle Stage Constants

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Defines the possible lifecycle stages for a data stream instance. These constants indicate whether an instance is in staging, production, or has been retired.

```go
const (
    LifeCycleStageStaging    = "staging"
    LifeCycleStageProduction = "production"
    LifeCycleStageRetired    = "retired"
)
```

--------------------------------

### Verify Channel Definitions

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/channel-definitions.md

Validates a set of channel definitions against available report codecs. Use this function to ensure channel configurations adhere to defined constraints before deployment.

```go
func VerifyChannelDefinitions(
    codecs map[llotypes.ReportFormat]ReportCodec,
    channelDefs llotypes.ChannelDefinitions,
) error
```

```go
error: ChannelDefinition with ID 1 has no streams
       ChannelDefinition with ID 2 has too many streams, got: 10001/10000
       invalid ChannelDefinition with ID 3: unknown report format
```

```go
codecs := map[llotypes.ReportFormat]llo.ReportCodec{
    llotypes.ReportFormatCapabilityTrigger: triggerCodec,
    llotypes.ReportFormatEVMABIEncoded: evmCodec,
}

defs := llotypes.ChannelDefinitions{
    1: channelDef1,
    2: channelDef2,
}

if err := llo.VerifyChannelDefinitions(codecs, defs); err != nil {
    log.Fatalf("Invalid channel definitions: %v", err)
}
```

--------------------------------

### gRPC Service Definition

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Defines the available RPC methods for the Transmitter service.

```protobuf
service Transmitter {
    rpc Transmit(TransmitRequest) returns (TransmitResponse);
    rpc LatestReport(LatestReportRequest) returns (LatestReportResponse);
}
```

--------------------------------

### Capability Trigger Format Options JSON

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/channel-definitions.md

This JSON snippet shows the format-specific options for the Capability Trigger format, detailing multipliers for a given stream ID.

```json
{
  "multipliers": [
    {
      "streamID": 1000000001,
      "multiplier": "10000"
    }
  ]
}
```

--------------------------------

### LLObservationProto Message Format

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Defines the protobuf message structure for LLO observations, including attested predecessor retirement, observation timestamp, and channel updates.

```protobuf
message LLOObservationProto {
    bytes attestedPredecessorRetirement = 1;
    bool shouldRetire = 2;
    int64 unixTimestampNanosecondsLegacy = 3; // legacy field
    uint64 unixTimestampNanoseconds = 7;      // modern field
    repeated uint32 removeChannelIDs = 4;
    map<uint32, LLOChannelDefinitionProto> updateChannelDefinitions = 5;
    map<uint32, LLOStreamValue> streamValues = 6;
}
```

--------------------------------

### Retirement Report Codec Interface

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Interface for serializing and deserializing retirement reports, used for predecessor handover.

```go
type RetirementReportCodec interface {
    Encode(rr RetirementReport) ([]byte, error)
    Decode(b []byte) (RetirementReport, error)
}
```

--------------------------------

### PluginFactory Type Definition

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/plugin-factory.md

Defines the structure of the PluginFactory and its associated parameters, including configuration, caches, data sources, and logging.

```go
type PluginFactory struct {
    PluginFactoryParams
}

type PluginFactoryParams struct {
    Config                               Config
    PredecessorRetirementReportCache    PredecessorRetirementReportCache
    ShouldRetireCache                   ShouldRetireCache
    RetirementReportCodec               RetirementReportCodec
    llotypes.ChannelDefinitionCache     ChannelDefinitionCache
    DataSource                          DataSource
    logger.Logger                       Logger
    OnchainConfigCodec                  OnchainConfigCodec
    ReportCodecs                        map[llotypes.ReportFormat]ReportCodec
    OutcomeTelemetryCh                  chan<- *LLOOutcomeTelemetry
    ReportTelemetryCh                   chan<- *LLOReportTelemetry
    DonID                               uint32
}
```

--------------------------------

### Encode Observation Method

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Encodes an Observation struct into a protobuf binary format. Compression can be enabled via the codec configuration.

```go
func (c protoObservationCodec) Encode(obs Observation) (types.Observation, error)
```

--------------------------------

### protoOutcomeCodecV0

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Implementation of OutcomeCodec for Version 0 (Legacy). Uses LLOOutcomeProtoV0 with specific timestamp and validity period handling.

```APIDOC
## Implementation: protoOutcomeCodecV0

### Description
Legacy implementation of the OutcomeCodec. Uses `LLOOutcomeProtoV0` and handles timestamps and validity periods differently from current versions.

- **ValidAfterSeconds** (int32) instead of nanoseconds
- **int64 observation timestamp**
```

--------------------------------

### LLOStreamsTriggerEvent and LLOStreamDecimal Output Structures

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/report-codecs.md

Defines the structure for the 'outputs' map within OCRTriggerReport, detailing how stream data and observation timestamps are represented.

```go
type LLOStreamsTriggerEvent struct {
    Payload []LLOStreamDecimal  // one per stream
    ObservationTimestampNanoseconds uint64
}

type LLOStreamDecimal struct {
    StreamID uint32
    Decimal []byte  // binary encoded decimal
}
```

--------------------------------

### Set Observation Timeout

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/README.md

Set an observation timeout using context.WithTimeout to manage data source latency. Ensure the cancel function is deferred to release resources.

```go
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
deferr cancel()
observation, err := plugin.Observation(ctx, outctx, nil)
```

--------------------------------

### protoOutcomeCodecV1

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Implementation of OutcomeCodec for Version 1 (Current). Uses LLOOutcomeProtoV1 with nanosecond precision for timestamps and validity periods.

```APIDOC
## Implementation: protoOutcomeCodecV1

### Description
Current implementation of the OutcomeCodec. Uses `LLOOutcomeProtoV1` for enhanced precision.

- **ValidAfterNanoseconds** (uint64) for nanosecond precision
- **uint64 observation timestamp**
```

--------------------------------

### EVMOnchainConfigCodec.Decode

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/codecs.md

Decodes on-chain configuration from a 64-byte EVM format. The format includes version and predecessor configuration digest. Validation checks ensure the byte length, version, and version fit within uint8.

```APIDOC
## EVMOnchainConfigCodec.Decode

### Description
Decodes on-chain configuration from 64-byte EVM format.

### Format
- Bytes 0-31: Version (big-endian uint256)
- Bytes 32-63: PredecessorConfigDigest (32-byte hash)

### Validation
- Length must be exactly 64 bytes
- Version must be 1
- Version must fit in uint8
```

--------------------------------

### Retrieve and Encode Observations

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/reporting-plugin.md

Retrieves observations from the data source and encodes them into a protobuf serialized format. Handles potential errors during observation, encoding, or channel definition updates. The query parameter is nil for LLO plugins.

```go
observation, err := plugin.Observation(ctx, outcomeContext, nil)
if err != nil {
    return fmt.Errorf("observation failed: %w", err)
}
// observation contains encoded stream values and channel updates
```

--------------------------------

### OptsCache Structure

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/types.md

Defines a cache for channel-specific options to prevent repeated deserialization. It includes methods for checking the cache size, resetting it, and removing specific channel options.

```go
type OptsCache struct {
    // Contains unexported fields
}
```

--------------------------------

### Transmit Reports Individually

Source: https://github.com/smartcontractkit/chainlink-data-streams/blob/master/_autodocs/endpoints.md

Iterate through a list of reports and transmit each one individually. Log any transmission failures encountered during the loop.

```go
for _, report := range reports {
    resp, err := client.Transmit(ctx, &rpc.TransmitRequest{
        Payload: report.Data,
        ReportFormat: report.Format,
    })
    if err != nil {
        log.Warnf("Transmit failed: %v", err)
    }
}
```