### Complete DRPC Server Example

Source: https://context7.com/storj/drpc/llms.txt

A full example demonstrating how to set up a DRPC server, register services, and listen for incoming connections.

```go
package main

import (
    "context"
    "net"

    "storj.io/drpc/drpcmux"
    "storj.io/drpc/drpcserver"

    "example.com/myservice/pb"
)

type CookieMonsterServer struct {
    pb.DRPCCookieMonsterUnimplementedServer
}

// EatCookie implements the RPC method
func (s *CookieMonsterServer) EatCookie(ctx context.Context, cookie *pb.Cookie) (*pb.Crumbs, error) {
    return &pb.Crumbs{Cookie: cookie}, nil
}

func main() {
    // Create the service implementation
    cookieMonster := &CookieMonsterServer{}

    // Create a drpc RPC mux
    m := drpcmux.New()

    // Register the service on the mux
    err := pb.DRPCRegisterCookieMonster(m, cookieMonster)
    if err != nil {
        panic(err)
    }

    // Create a drpc server
    s := drpcserver.New(m)

    // Listen on a tcp socket
    lis, err := net.Listen("tcp", ":8080")
    if err != nil {
        panic(err)
    }

    // Run the server
    // For TLS, wrap the net.Listener with TLS before passing to Serve
    err = s.Serve(context.Background(), lis)
    if err != nil {
        panic(err)
    }
}
```

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

### DRPC with HTTP Server Example

Source: https://context7.com/storj/drpc/llms.txt

An example demonstrating how to set up a DRPC server with an HTTP handler that supports JSON and protobuf, alongside a gRPC migration listener.

```APIDOC
## DRPC with HTTP Server Example

This example shows how to create a DRPC server using `drpcserver` and expose it via HTTP using `drpchttp.New`. It also utilizes `drpcmigrate.NewListenMux` to allow DRPC and HTTP traffic on the same port.

### Server Setup

1.  Define your RPC service (e.g., `CookieMonsterServer`).
2.  Create a DRPC multiplexer (`drpcmux.New`) and register your service.
3.  Listen on a TCP socket.
4.  Create a `drpcmigrate.ListenMux` to route traffic based on headers.
5.  Start a DRPC server on the DRPC-routed listener.
6.  Start an HTTP server using `drpchttp.New` on the default listener.
7.  Run the `ListenMux` to handle incoming connections.

```go
package main

import (
    "context"
    "net"
    "net/http"

    "golang.org/x/sync/errgroup"

    "storj.io/drpc/drpchttp"
    "storj.io/drpc/drpcmigrate"
    "storj.io/drpc/drpcmux"
    "storj.io/drpc/drpcserver"

    "example.com/myservice/pb"
)

type CookieMonsterServer struct {
    pb.DRPCCookieMonsterUnimplementedServer
}

func (s *CookieMonsterServer) EatCookie(ctx context.Context, cookie *pb.Cookie) (*pb.Crumbs, error) {
    return &pb.Crumbs{Cookie: cookie}, nil
}

func main() {
    cookieMonster := &CookieMonsterServer{}

    // Create and register the mux
    m := drpcmux.New()
    pb.DRPCRegisterCookieMonster(m, cookieMonster)

    // Listen on a tcp socket
    lis, _ := net.Listen("tcp", ":8080")

    // Create a listen mux to route DRPC and HTTP traffic
    lisMux := drpcmigrate.NewListenMux(lis, len(drpcmigrate.DRPCHeader))
    drpcLis := lisMux.Route(drpcmigrate.DRPCHeader)
    httpLis := lisMux.Default()

    var group errgroup.Group

    // DRPC handling
    group.Go(func() error {
        s := drpcserver.New(m)
        return s.Serve(context.Background(), drpcLis)
    })

    // HTTP handling (supports JSON and protobuf)
    group.Go(func() error {
        s := http.Server{Handler: drpchttp.New(m)}
        return s.Serve(httpLis)
    })

    // Run the listen mux
    group.Go(func() error {
        return lisMux.Run(context.Background())
    })

    group.Wait()
}
```
```

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

### DRPC with HTTP Server Example

Source: https://context7.com/storj/drpc/llms.txt

This example demonstrates setting up a DRPC server and an HTTP server that handles DRPC requests using `drpchttp.New`. It uses `drpcmux` for routing and `drpcmigrate.NewListenMux` to manage both DRPC and HTTP traffic on the same listener.

```go
package main

import (
    "context"
    "net"
    "net/http"

    "golang.org/x/sync/errgroup"

    "storj.io/drpc/drpchttp"
    "storj.io/drpc/drpcmigrate"
    "storj.io/drpc/drpcmux"
    "storj.io/drpc/drpcserver"

    "example.com/myservice/pb"
)

type CookieMonsterServer struct {
    pb.DRPCCookieMonsterUnimplementedServer
}

func (s *CookieMonsterServer) EatCookie(ctx context.Context, cookie *pb.Cookie) (*pb.Crumbs, error) {
    return &pb.Crumbs{Cookie: cookie}, nil
}

func main() {
    cookieMonster := &CookieMonsterServer{}

    // Create and register the mux
    m := drpcmux.New()
    pb.DRPCRegisterCookieMonster(m, cookieMonster)

    // Listen on a tcp socket
    lis, _ := net.Listen("tcp", ":8080")

    // Create a listen mux to route DRPC and HTTP traffic
    lisMux := drpcmigrate.NewListenMux(lis, len(drpcmigrate.DRPCHeader))
    drpcLis := lisMux.Route(drpcmigrate.DRPCHeader)
    httpLis := lisMux.Default()

    var group errgroup.Group

    // DRPC handling
    group.Go(func() error {
        s := drpcserver.New(m)
        return s.Serve(context.Background(), drpcLis)
    })

    // HTTP handling (supports JSON and protobuf)
    group.Go(func() error {
        s := http.Server{Handler: drpchttp.New(m)}
        return s.Serve(httpLis)
    })

    // Run the listen mux
    group.Go(func() error {
        return lisMux.Run(context.Background())
    })

    group.Wait()
}
```

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

### Install and Use protoc-gen-go-drpc

Source: https://context7.com/storj/drpc/llms.txt

Installs the DRPC protocol buffer plugin and generates DRPC client/server code from proto files.

```bash
# Install the protoc plugin
go install storj.io/drpc/cmd/protoc-gen-go-drpc@latest

# Generate DRPC code from proto files
protoc --go_out=. --go-drpc_out=. service.proto
```

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

### Tracker Example

Source: https://context7.com/storj/drpc/llms.txt

An example demonstrating the usage of drpcctx.Tracker to manage and wait for multiple goroutines to complete.

```APIDOC
## Tracker Example
This example demonstrates how to use the `drpcctx.Tracker` to launch and manage multiple goroutines.

### Usage:
1. Create a context with a timeout.
2. Initialize a `drpcctx.NewTracker` with the context.
3. Use `tracker.Run` to launch goroutines, passing a callback function that accepts a context.
4. Inside the callback, use `select` to handle context cancellation or perform work.
5. Call `tracker.Wait()` to block until all launched goroutines have finished.
```

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

### Manual Flush Example

Source: https://context7.com/storj/drpc/llms.txt

Example demonstrating how to use manual flushing to batch messages before sending them over a DRPC stream.

```APIDOC
## Manual Flush Example
This example shows how to batch messages using manual flush control on a DRPC stream.

### Function: sendBatchedMessages
- **stream** (drpc.Stream) - The DRPC stream to send messages on.
- **messages** ([]*Message) - A slice of messages to send.

### Logic:
1. Obtain the underlying stream with manual flush control.
2. Disable auto-flush to enable batching.
3. Iterate through messages, sending each one.
4. Re-enable auto-flush for the last message.
5. Return any errors encountered during sending.
```

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

### HTTP Client Example

Source: https://context7.com/storj/drpc/llms.txt

An example of how to make an HTTP POST request to a DRPC service endpoint using JSON encoding.

```APIDOC
## HTTP Client Example

This example demonstrates how to send a JSON-encoded request to a DRPC service endpoint exposed over HTTP.

### Request Details

- **Method**: POST
- **Endpoint**: `http://localhost:8080/sesamestreet.CookieMonster/EatCookie`
- **Content-Type**: `application/json`
- **Request Body**: A JSON object representing the `Cookie` message (e.g., `{"type": "Chocolate"}`).

### Response Handling

- The response is expected to be JSON.
- The example decodes the response into a struct and prints the cookie type.

```go
package main

import (
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
)

func main() {
    // Make HTTP request to DRPC service (JSON)
    resp, err := http.Post(
        "http://localhost:8080/sesamestreet.CookieMonster/EatCookie",
        "application/json",
        strings.NewReader(`{"type": "Chocolate"}`),
    )
    if err != nil {
        panic(err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        panic(fmt.Sprintf("unexpected status: %d", resp.StatusCode))
    }

    var data struct {
        Cookie struct {
            Type string `json:"type"`
        } `json:"cookie"`
    }
    json.NewDecoder(resp.Body).Decode(&data)
    fmt.Println(data.Cookie.Type) // Output: Chocolate
}
```
```

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

### DRPC Metadata Example

Source: https://context7.com/storj/drpc/llms.txt

Illustrates how to add metadata (single key-value pairs and multiple pairs) to a context and then retrieve it.

```go
package main

import (
    "context"
    "fmt"

    "storj.io/drpc/drpcmetadata"
)

func main() {
    // Add metadata to context
    ctx := context.Background()
    ctx = drpcmetadata.Add(ctx, "user-id", "12345")
    ctx = drpcmetadata.Add(ctx, "request-id", "abc-123")

    // Or add multiple pairs at once
    ctx = drpcmetadata.AddPairs(ctx, map[string]string{
        "auth-token": "secret-token",
        "client-version": "1.0.0",
    })

    // Retrieve metadata from context
    metadata, ok := drpcmetadata.Get(ctx)
    if ok {
        fmt.Println(metadata["user-id"])
        fmt.Println(metadata["request-id"])
    }
}
```

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

### Example DRPC Proto File

Source: https://context7.com/storj/drpc/llms.txt

Defines a sample service and messages for use with DRPC code generation.

```protobuf
syntax = "proto3";
option go_package = "example.com/myservice/pb";

package myservice;

message Cookie {
  enum Type {
    Sugar = 0;
    Oatmeal = 1;
    Chocolate = 2;
  }
  Type type = 1;
}

message Crumbs {
  Cookie cookie = 1;
}

service CookieMonster {
  rpc EatCookie(Cookie) returns (Crumbs) {}
}
```

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

### OpenTelemetry Tracing Middleware in Go

Source: https://context7.com/storj/drpc/llms.txt

Provides an example of creating middleware for OpenTelemetry integration. It extracts trace context from metadata, starts a new span, and wraps the stream with a context containing the span. This allows for distributed tracing of RPC calls.

```go
package main

import (
    "context"
    "log"
    "time"

    "storj.io/drpc"
)

// Example with OpenTelemetry tracing
type OTelHandler struct {
    handler drpc.Handler
}

func (h *OTelHandler) HandleRPC(stream drpc.Stream, rpc string) error {
    metadata, ok := drpcmetadata.Get(stream.Context())
    if ok {
        // Extract trace context from metadata
        ctx := otel.GetTextMapPropagator().Extract(
            stream.Context(), propagation.MapCarrier(metadata))
        ctx, span := tracer.Start(ctx, "HandleRPC")
        defer span.End()

        // Wrap stream with new context
        stream = &streamWrapper{Stream: stream, ctx: ctx}
    }
    return h.handler.HandleRPC(stream, rpc)
}

type streamWrapper struct {
    drpc.Stream
    ctx context.Context
}

func (s *streamWrapper) Context() context.Context { return s.ctx }
```

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

### Complete DRPC Client Example

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates how to establish a DRPC client connection, create a service-specific client, and execute an RPC call with a timeout.

```go
package main

import (
    "context"
    "fmt"
    "net"
    "time"

    "storj.io/drpc/drpcconn"

    "example.com/myservice/pb"
)

func main() {
    // Dial the drpc server
    rawconn, err := net.Dial("tcp", "localhost:8080")
    if err != nil {
        panic(err)
    }
    // For TLS, wrap the net.Conn with TLS before making a DRPC conn

    // Convert the net.Conn to a drpc.Conn
    conn := drpcconn.New(rawconn)
    defer conn.Close()

    // Create a proto-specific client
    client := pb.NewDRPCCookieMonsterClient(conn)

    // Set a deadline for the operation
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()

    // Execute the RPC
    crumbs, err := client.EatCookie(ctx, &pb.Cookie{
        Type: pb.Cookie_Oatmeal,
    })
    if err != nil {
        panic(err)
    }

    fmt.Println(crumbs.Cookie.Type.String()) // Output: Oatmeal
}
```

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

### HTTP Client Example for DRPC Service

Source: https://context7.com/storj/drpc/llms.txt

This example shows how to make an HTTP POST request to a DRPC service endpoint using JSON encoding. It demonstrates sending a JSON payload and handling the JSON response.

```go
package main

import (
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
)

func main() {
    // Make HTTP request to DRPC service (JSON)
    resp, err := http.Post(
        "http://localhost:8080/sesamestreet.CookieMonster/EatCookie",
        "application/json",
        strings.NewReader(`{"type": "Chocolate"}`),
    )
    if err != nil {
        panic(err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK {
        panic(fmt.Sprintf("unexpected status: %d", resp.StatusCode))
    }

    var data struct {
        Cookie struct {
            Type string `json:"type"`
        } `json:"cookie"`
    }
    json.NewDecoder(resp.Body).Decode(&data)
    fmt.Println(data.Cookie.Type) // Output: Chocolate
}
```

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

### DRPC Tracker Example

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates using the DRPC context tracker to launch multiple goroutines, wait for their completion, and handle context cancellation.

```go
package main

import (
    "context"
    "fmt"
    "time"

    "storj.io/drpc/drpcctx"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    tracker := drpcctx.NewTracker(ctx)

    // Launch multiple goroutines
    for i := 0; i < 3; i++ {
        id := i
        tracker.Run(func(ctx context.Context) {
            select {
            case <-ctx.Done():
                fmt.Printf("Worker %d cancelled\n", id)
            case <-time.After(time.Second):
                fmt.Printf("Worker %d completed\n", id)
            }
        })
    }

    // Wait for all goroutines to complete
    tracker.Wait()
    fmt.Println("All workers done")
}
```

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

### Get Chan Instance

Source: https://github.com/storj/drpc/blob/main/drpcsignal/README.md

Returns the channel, allocating it if necessary. Use this to access the underlying channel.

```go
func (c *Chan) Get() chan struct{} {
}

```

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

### ListenMux Run Method

Source: https://github.com/storj/drpc/blob/main/drpcmigrate/README.md

Run starts the ListenMux's operation. It listens on the provided base listener and dispatches accepted connections to the appropriate routed listeners based on their prefixes.

```go
func (m *ListenMux) Run(ctx context.Context) error
```

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

### DRPC Connection Pool Example

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates how to create and use a DRPC connection pool with custom options for expiration and capacity limits. Includes dialing logic for new connections.

```go
package main

import (
    "context"
    "net"
    "time"

    "storj.io/drpc/drpcconn"
    "storj.io/drpc/drpcpool"
)

func main() {
    // Create a connection pool with expiration and capacity limits
    pool := drpcpool.New[string, *drpcconn.Conn](drpcpool.Options{
        Expiration:  5 * time.Minute,  // Connections expire after 5 minutes
        Capacity:    100,              // Maximum 100 total connections
        KeyCapacity: 10,               // Maximum 10 connections per key
    })
    defer pool.Close()

    // Get a connection from the pool (will dial if needed)
    conn := pool.Get(context.Background(), "localhost:8080",
        func(ctx context.Context, key string) (*drpcconn.Conn, error) {
            rawconn, err := net.Dial("tcp", key)
            if err != nil {
                return nil, err
            }
            return drpcconn.New(rawconn), nil
        })
    defer conn.Close()

    // Use the connection...
}
```

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

### Get All Metadata from Context

Source: https://github.com/storj/drpc/blob/main/drpcmetadata/README.md

Use Get to retrieve all key/value pairs associated with the given context. It returns the metadata map and a boolean indicating if any metadata was found.

```go
func Get(ctx context.Context) (map[string]string, bool)
```

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

### DRPC Error Handling Example

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates how to create errors with specific DRPC codes and how to retrieve these codes from an error object.

```go
package main

import (
    "errors"
    "fmt"

    "storj.io/drpc/drpcerr"
)

const (
    CodeNotFound     = 1
    CodeUnauthorized = 2
    CodeInvalidInput = 3
)

func processRequest(id string) error {
    if id == "" {
        return drpcerr.WithCode(errors.New("id is required"), CodeInvalidInput)
    }
    if id == "unknown" {
        return drpcerr.WithCode(errors.New("resource not found"), CodeNotFound)
    }
    return nil
}

func main() {
    err := processRequest("")
    if err != nil {
        code := drpcerr.Code(err)
        fmt.Printf("Error code: %d, message: %v\n", code, err)
    }
}
```

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

### Stream Cache Example in Go

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates how to use drpccache to load or create expensive resources once per stream. The LoadOrCreate function ensures that the provided function is only called once for a given key within the same stream context.

```go
package main

import (
    "context"

    "storj.io/drpc/drpccache"
)

func handleRPC(ctx context.Context) {
    cache := drpccache.FromContext(ctx)
    if cache == nil {
        return
    }

    // Load or create expensive resource once per stream
    dbConn := cache.LoadOrCreate("db-connection", func() interface{} {
        // This will only be called once per stream
        return connectToDatabase()
    })

    // Use the cached connection
    _ = dbConn
}

func connectToDatabase() interface{} {
    // Expensive database connection setup
    return "db-connection"
}
```

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

### Protocol Buffers Code Generation with protoc-gen-go-drpc

Source: https://context7.com/storj/drpc/llms.txt

Instructions and examples for using the `protoc-gen-go-drpc` plugin to generate DRPC client and server code from Protocol Buffer definitions.

```APIDOC
## Protocol Buffers Code Generation

### protoc-gen-go-drpc
Generates DRPC client and server code from Protocol Buffer service definitions. Install and use with protoc.

```bash
# Install the protoc plugin
go install storj.io/drpc/cmd/protoc-gen-go-drpc@latest

# Generate DRPC code from proto files
protoc --go_out=. --go-drpc_out=. service.proto
```

Example proto file:

```protobuf
syntax = "proto3";
option go_package = "example.com/myservice/pb";

package myservice;

message Cookie {
  enum Type {
    Sugar = 0;
    Oatmeal = 1;
    Chocolate = 2;
  }
  Type type = 1;
}

message Crumbs {
  Cookie cookie = 1;
}

service CookieMonster {
  rpc EatCookie(Cookie) returns (Crumbs) {}
}
```
```

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

### ListenMux Route Method

Source: https://github.com/storj/drpc/blob/main/drpcmigrate/README.md

Route registers a new listener for a specific prefix. Connections starting with this prefix will be directed to the returned listener. The prefix length must match the ListenMux's configured prefixLen.

```go
func (m *ListenMux) Route(prefix string) net.Listener
```

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

### DRPC Manual Flush Example

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates how to manually control flushing for batching messages on a DRPC stream by disabling auto-flush and re-enabling it for the last message.

```go
package main

import "storj.io/drpc"

func sendBatchedMessages(stream drpc.Stream, messages []*Message) error {
    // Get the underlying stream for manual flush control
    flusher := stream.(interface {
        GetStream() drpc.Stream
    }).GetStream().(interface {
        SetManualFlush(bool)
    })

    // Disable auto-flush for batching
    flusher.SetManualFlush(true)

    for i, msg := range messages {
        // Re-enable auto-flush for last message
        if i == len(messages)-1 {
            flusher.SetManualFlush(false)
        }
        if err := stream.Send(msg); err != nil {
            return err
        }
    }
    return nil
}
```

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

### Run Callback in Goroutine with Tracker

Source: https://github.com/storj/drpc/blob/main/drpcctx/README.md

Run starts a new goroutine that executes the provided callback function. The callback receives the tracker's context, allowing it to be managed by the tracker.

```go
func (t *Tracker) Run(cb func(ctx context.Context))
```

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

### Get Context from HTTP Request

Source: https://github.com/storj/drpc/blob/main/drpchttp/README.md

Retrieves the context.Context from an http.Request, applying any metadata from the X-Drpc-Metadata header.

```go
func Context(req *http.Request) (context.Context, error)
```

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

### Get Cache from Context

Source: https://github.com/storj/drpc/blob/main/drpccache/README.md

Retrieve a cache instance associated with the provided context. Returns nil if no cache is found.

```go
cache := drpccache.FromContext(stream.Context())
if cache != nil {
       value := cache.LoadOrCreate("initialized", func() (interface{}) {
               return 42
       })
}
```

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

### Construct New Server With Options

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

NewWithOptions constructs a new Server using provided options to tune connection handling. Ensure options are correctly configured before use.

```go
func NewWithOptions(handler drpc.Handler, opts Options) *Server
```

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

### Define Server Options

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

Options controls configuration settings for a server. Use this to manage managers, logging, and statistics collection.

```go
type Options struct {
	// Manager controls the options we pass to the managers this server creates.
	Manager drpcmanager.Options

	// Log is called when errors happen that can not be returned up, like
	// temporary network errors when accepting connections, or errors
	// handling individual clients. It is not called if nil.
	Log func(error)

	// CollectStats controls whether the server should collect stats on the
	// rpcs it serves.
	CollectStats bool
}
```

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

### Get Stream ID

Source: https://github.com/storj/drpc/blob/main/drpcstream/README.md

Returns the unique identifier for this stream.

```go
func (s *Stream) ID() uint64
```

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

### Get drpcconn.Conn Transport

Source: https://github.com/storj/drpc/blob/main/drpcconn/README.md

Returns the transport used by the connection.

```go
func (c *Conn) Transport() drpc.Transport
```

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

### Get drpcconn.Conn Stats

Source: https://github.com/storj/drpc/blob/main/drpcconn/README.md

Returns collected stats grouped by RPC.

```go
func (c *Conn) Stats() map[string]drpcstats.Stats
```

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

### Construct New Server

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

New constructs a new Server. It requires a drpc.Handler to process requests.

```go
func New(handler drpc.Handler) *Server
```

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

### DRPC Client Dialing with Headers in Go

Source: https://context7.com/storj/drpc/llms.txt

Demonstrates three methods for establishing a DRPC connection from a client, all ensuring the correct DRPC header is sent. Method 1 uses DialWithHeader, Method 2 uses HeaderDialer, and Method 3 wraps an existing net.Conn.

```go
package main

import (
    "context"
    "net"

    "storj.io/drpc/drpcconn"
    "storj.io/drpc/drpcmigrate"
)

func main() {
    // Method 1: Use DialWithHeader
    rawconn, err := drpcmigrate.DialWithHeader(
        context.Background(), "tcp", "localhost:8080", drpcmigrate.DRPCHeader)
    if err != nil {
        panic(err)
    }

    // Method 2: Use HeaderDialer
    dialer := &drpcmigrate.HeaderDialer{Header: drpcmigrate.DRPCHeader}
    rawconn, err = dialer.DialContext(context.Background(), "tcp", "localhost:8080")
    if err != nil {
        panic(err)
    }

    // Method 3: Wrap existing connection
    rawconn, _ = net.Dial("tcp", "localhost:8080")
    rawconn = drpcmigrate.NewHeaderConn(rawconn, drpcmigrate.DRPCHeader)

    // Create DRPC connection
    conn := drpcconn.New(rawconn)
    defer conn.Close()
}
```

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

### Get drpcconn.Conn Closed Channel

Source: https://github.com/storj/drpc/blob/main/drpcconn/README.md

Returns a channel that is closed when the connection is closed.

```go
func (c *Conn) Closed() <-chan struct{}
```

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

### Get DRPC Server Statistics

Source: https://context7.com/storj/drpc/llms.txt

Retrieves collected statistics for the DRPC server, grouped by RPC.

```go
// Stats returns collected stats grouped by rpc.
func (s *Server) Stats() map[string]drpcstats.Stats
```

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

### Register Service with Description

Source: https://github.com/storj/drpc/blob/main/drpcmux/README.md

Register associates the RPCs described by the description in the server. It returns an error if there was a problem registering it.

```go
func (m *Mux) Register(srv interface{}, desc drpc.Description) error
```

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

### Create New Mux Instance

Source: https://github.com/storj/drpc/blob/main/drpcmux/README.md

New constructs a new Mux. Use this to initialize a Mux handler for dispatching RPCs.

```go
func New() *Mux
```

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

### Get Closed Channel for Manager in drpcmanager

Source: https://github.com/storj/drpc/blob/main/drpcmanager/README.md

Obtain a channel that will be closed when the Manager is shut down. This is useful for signaling or synchronization.

```go
func (m *Manager) Closed() <-chan struct{}
```

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

### Initialize Chan with Capacity

Source: https://github.com/storj/drpc/blob/main/drpcsignal/README.md

Sets the channel to a freshly allocated channel with the provided capacity. This method is a no-op if called after any other methods on the Chan.

```go
func (c *Chan) Make(cap uint) {
}

```

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

### Get Channel Closed on Signal Set

Source: https://github.com/storj/drpc/blob/main/drpcsignal/README.md

Returns a channel that will be closed when the signal is set. This can be used for waiting or notification.

```go
func (s *Signal) Signal() chan struct{} {
}

```

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

### DRPC Migration Tools

Source: https://context7.com/storj/drpc/llms.txt

The `drpcmigrate` package provides tools to run DRPC alongside gRPC on the same ports. `NewListenMux` creates a listener that can route connections based on a prefix, allowing separate listeners for DRPC and other traffic.

```go
import "storj.io/drpc/drpcmigrate"

// DRPCHeader is the header for DRPC connections.
var DRPCHeader = "DRPC!!!1"

// NewListenMux creates a ListenMux that reads prefixLen bytes from connections.
func NewListenMux(base net.Listener, prefixLen int) *ListenMux

// Route returns a listener for connections matching the prefix.
func (m *ListenMux) Route(prefix string) net.Listener

// Default returns the listener for connections that don't match any route.
func (m *ListenMux) Default() net.Listener

// Run accepts connections and routes them appropriately.
func (m *ListenMux) Run(ctx context.Context) error

// DialWithHeader dials with a custom header prefix.
func DialWithHeader(ctx context.Context, network, address, header string) (net.Conn, error)

// NewHeaderConn wraps a conn to write a header on first write.
func NewHeaderConn(conn net.Conn, header string) *HeaderConn
```

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

### Get Server Statistics

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

Stats returns the collected statistics, grouped by rpc. This is useful for monitoring server performance.

```go
func (s *Server) Stats() map[string]drpcstats.Stats
```

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

### Get Error Code

Source: https://github.com/storj/drpc/blob/main/drpcerr/README.md

Retrieves the error code associated with a given error. Returns 0 if no code is found.

```go
func Code(err error) uint64
```

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

### Create New drpcconn.Conn With Options

Source: https://github.com/storj/drpc/blob/main/drpcconn/README.md

Creates a new drpc client connection with configurable options.

```go
func NewWithOptions(tr drpc.Transport, opts Options) *Conn
```

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

### Close Tracker and Goroutines

Source: https://github.com/storj/drpc/blob/main/drpctest/README.md

Close cancels the tracker's context and ensures all goroutines started with Run have completed.

```go
func (t *Tracker) Close()
```

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

### Create New Cache

Source: https://github.com/storj/drpc/blob/main/drpccache/README.md

Instantiate a new, empty cache. This cache is not associated with any context by default.

```go
func New() *Cache {
}
```

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

### Get Stream Context

Source: https://github.com/storj/drpc/blob/main/drpcstream/README.md

Returns the context associated with the stream. This context is closed when the stream can no longer issue reads or writes.

```go
func (s *Stream) Context() context.Context
```

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

### Get String Representation of Manager in drpcmanager

Source: https://github.com/storj/drpc/blob/main/drpcmanager/README.md

Retrieve a string representation of the Manager. This is typically used for debugging or logging purposes.

```go
func (m *Manager) String() string
```

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

### Create New Manager with Options in drpcmanager

Source: https://github.com/storj/drpc/blob/main/drpcmanager/README.md

Create a new Manager with custom options to control its behavior. Options allow fine-tuning of buffer sizes, reader/stream configurations, and cancellation strategies.

```go
func NewWithOptions(tr drpc.Transport, opts Options) *Manager
```

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

### Get drpc Transport from Context

Source: https://github.com/storj/drpc/blob/main/drpcctx/README.md

Use Transport to retrieve the drpc.Transport associated with a context. It returns the transport and a boolean indicating its presence.

```go
func Transport(ctx context.Context) (drpc.Transport, bool)
```

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

### Create New Pool

Source: https://github.com/storj/drpc/blob/main/drpcpool/README.md

Constructs a new Pool with the specified Options. This function initializes the pool's internal structures based on the provided configuration.

```go
func New[K comparable, V Conn](opts Options) *Pool[K, V]
```

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

### Get Signal Error and Validity

Source: https://github.com/storj/drpc/blob/main/drpcsignal/README.md

Returns the error set with the signal and a boolean indicating if the result is valid. The boolean is true if an error has been set.

```go
func (s *Signal) Get() (error, bool) {
}

```

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

### Create HTTP Handler with Options

Source: https://github.com/storj/drpc/blob/main/drpchttp/README.md

Returns a net/http.Handler that dispatches to a drpc.Handler, hosting RPCs at paths like /service.Server/Method. Supports metadata via X-Drpc-Metadata headers and various content types for different protocols.

```go
func NewWithOptions(handler drpc.Handler, os ...Option) http.Handler
```

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

### Serve Drpc Requests

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

Serve listens for connections on the listener and serves drpc requests on new connections. It requires a context and a net.Listener.

```go
func (s *Server) Serve(ctx context.Context, lis net.Listener) (err error)
```

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

### Get DRPC Client Statistics and Transport

Source: https://context7.com/storj/drpc/llms.txt

Retrieves collected statistics for a DRPC client connection, grouped by RPC, and provides access to the underlying transport.

```go
// Stats returns collected stats grouped by rpc.
func (c *Conn) Stats() map[string]drpcstats.Stats
```

```go
// Transport returns the transport the conn is using.
func (c *Conn) Transport() drpc.Transport
```

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

### DRPC Server Implementation

Source: https://context7.com/storj/drpc/llms.txt

Provides details on creating and running a DRPC server, including configuration options and core methods.

```APIDOC
## drpcserver - Server Implementation

Package drpcserver allows execution of registered RPCs.

### Server Configuration Options

```go
type Options struct {
    Manager      drpcmanager.Options  // Manager configuration
    Log          func(error)          // Error logging callback
    CollectStats bool                 // Enable stats collection
}
```

### Core Methods

- **New(handler drpc.Handler) *Server**: Constructs a new Server.
- **NewWithOptions(handler drpc.Handler, opts Options) *Server**: Constructs a new Server with custom options.
- **Serve(ctx context.Context, lis net.Listener) error**: Listens for connections and serves drpc requests.
- **ServeOne(ctx context.Context, tr drpc.Transport) error**: Serves a single set of rpcs on the provided transport.
- **Stats() map[string]drpcstats.Stats**: Returns collected stats grouped by rpc.

### Complete Server Example

```go
package main

import (
    "context"
    "net"

    "storj.io/drpc/drpcmux"
    "storj.io/drpc/drpcserver"

    "example.com/myservice/pb"
)

type CookieMonsterServer struct {
    pb.DRPCCookieMonsterUnimplementedServer
}

// EatCookie implements the RPC method
func (s *CookieMonsterServer) EatCookie(ctx context.Context, cookie *pb.Cookie) (*pb.Crumbs, error) {
    return &pb.Crumbs{Cookie: cookie}, nil
}

func main() {
    // Create the service implementation
    cookieMonster := &CookieMonsterServer{}

    // Create a drpc RPC mux
    m := drpcmux.New()

    // Register the service on the mux
    err := pb.DRPCRegisterCookieMonster(m, cookieMonster)
    if err != nil {
        panic(err)
    }

    // Create a drpc server
    s := drpcserver.New(m)

    // Listen on a tcp socket
    lis, err := net.Listen("tcp", ":8080")
    if err != nil {
        panic(err)
    }

    // Run the server
    // For TLS, wrap the net.Listener with TLS before passing to Serve
    err = s.Serve(context.Background(), lis)
    if err != nil {
        panic(err)
    }
}
```
```

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

### Create and Serve DRPC Requests

Source: https://context7.com/storj/drpc/llms.txt

Provides functions to create a new DRPC server with or without custom options, and to serve RPC requests over a listener.

```go
// New constructs a new Server.
func New(handler drpc.Handler) *Server
```

```go
// NewWithOptions constructs a new Server with custom options.
func NewWithOptions(handler drpc.Handler, opts Options) *Server
```

```go
// Serve listens for connections and serves drpc requests.
func (s *Server) Serve(ctx context.Context, lis net.Listener) error
```

```go
// ServeOne serves a single set of rpcs on the provided transport.
func (s *Server) ServeOne(ctx context.Context, tr drpc.Transport) error
```

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

### DRPC Connection Pool Operations

Source: https://context7.com/storj/drpc/llms.txt

Provides essential methods for managing a DRPC connection pool, including creating, getting, putting, taking, and closing connections.

```go
// New constructs a new Pool with the provided Options.
func New[K comparable, V Conn](opts Options) *Pool[K, V]

// Get returns a Conn that uses the dial function to create underlying connections.
func (p *Pool[K, V]) Get(ctx context.Context, key K,
    dial func(ctx context.Context, key K) (V, error)) Conn

// Put places the connection into the cache with the provided key.
func (p *Pool[K, V]) Put(key K, val V)

// Take acquires a value from the cache if one exists.
func (p *Pool[K, V]) Take(key K) (V, bool)

// Close evicts all entries from the Pool's cache.
func (p *Pool[K, V]) Close() error
```

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

### Get drpcconn.Conn Unblocked Channel

Source: https://github.com/storj/drpc/blob/main/drpcconn/README.md

Returns a channel closed when the connection is no longer blocked by a canceled Invoke or NewStream call. Do not call concurrently with Invoke or NewStream.

```go
func (c *Conn) Unblocked() <-chan struct{}
```

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

### Get Stored Error from Signal

Source: https://github.com/storj/drpc/blob/main/drpcsignal/README.md

Returns the error stored in the signal. Care must be taken as a nil error can be stored. A non-nil error indicates the Signal has been set.

```go
func (s *Signal) Err() error {
}

```

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

### Construct DRPC Reader with Options

Source: https://github.com/storj/drpc/blob/main/drpcwire/README.md

Use NewReaderWithOptions to create a Reader for Packets from an io.Reader, managing buffering with provided options.

```go
func NewReaderWithOptions(r io.Reader, opts ReaderOptions) *Reader
```

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

### Wait for All Goroutines to Complete

Source: https://github.com/storj/drpc/blob/main/drpcctx/README.md

Wait blocks the current goroutine until all callbacks that were started with the Run method have finished execution. This is useful for ensuring all background tasks are complete before proceeding.

```go
func (t *Tracker) Wait()
```

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

### Load or Create Value in Cache

Source: https://github.com/storj/drpc/blob/main/drpccache/README.md

Attempt to load a value by key. If the key does not exist, execute the provided function to create and store the value before returning it.

```go
func (cache *Cache) LoadOrCreate(key interface{}, fn func() interface{}) interface{} {
}
```

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

### Take Connection from Pool

Source: https://github.com/storj/drpc/blob/main/drpcpool/README.md

Acquires a connection from the cache if one is available for the given key. Returns the zero value for the connection type and false if no connection is found.

```go
func (p *Pool[K, V]) Take(key K) (V, bool)
```

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

### gRPC Service Server Interface (Go)

Source: https://github.com/storj/drpc/wiki/Protoc-output:-Go

Defines the server interface for a gRPC service. Implement this to create a gRPC server. Includes a base implementation for forward compatibility.

```go
// ServiceServer is the server API for Service service.
// All implementations must embed UnimplementedServiceServer
// for forward compatibility
type ServiceServer interface {
	Method1(context.Context, *In) (*Out, error)
	Method2(Service_Method2Server) error
	Method3(*In, Service_Method3Server) error
	Method4(Service_Method4Server) error
	mustEmbedUnimplementedServiceServer()
}

// UnimplementedServiceServer must be embedded to have forward compatible implementations.
type UnimplementedServiceServer struct {
}

func (UnimplementedServiceServer) Method1(context.Context, *In) (*Out, error) {
	return nil, status.Errorf(codes.Unimplemented, "method Method1 not implemented")
}
func (UnimplementedServiceServer) Method2(Service_Method2Server) error {
	return status.Errorf(codes.Unimplemented, "method Method2 not implemented")
}
func (UnimplementedServiceServer) Method3(*In, Service_Method3Server) error {
	return status.Errorf(codes.Unimplemented, "method Method3 not implemented")
}
func (UnimplementedServiceServer) Method4(Service_Method4Server) error {
	return status.Errorf(codes.Unimplemented, "method Method4 not implemented")
}
func (UnimplementedServiceServer) mustEmbedUnimplementedServiceServer() {}

```

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

### Unimplemented DRPCServiceServer Method1

Source: https://github.com/storj/drpc/blob/main/internal/integration/README.md

Default implementation for Method1 on DRPCServiceUnimplementedServer. It should be overridden by concrete server implementations.

```go
func (s *DRPCServiceUnimplementedServer) Method1(context.Context, *In) (*Out, error)
```

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

### Create New Server Stream with drpcmanager

Source: https://github.com/storj/drpc/blob/main/drpcmanager/README.md

Start a new server stream on the managed transport, waiting for an incoming invoke message from a client. This method is used by servers to accept requests.

```go
func (m *Manager) NewServerStream(ctx context.Context) (stream *drpcstream.Stream, rpc string, err error)
```

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

### Define Stream Options

Source: https://github.com/storj/drpc/blob/main/drpcstream/README.md

Options controls configuration settings for a stream. Use SplitSize to control packet frame splitting and ManualFlush to control automatic flushing.

```go
type Options struct {
	// SplitSize controls the default size we split packets into frames.
	SplitSize int

	// ManualFlush controls if the stream will automatically flush after every
	// message send. Note that flushing is not part of the drpc.Stream
	// interface, so if you use this you must be ready to type assert and call
	// RawFlush dynamically.
	ManualFlush bool

	// MaximumBufferSize causes the Stream to drop any internal buffers that are
	// larger than this amount to control maximum memory usage at the expense of
	// more allocations. 0 is unlimited.
	MaximumBufferSize int

	// Internal contains options that are for internal use only.
	Internal drpcopts.Stream
}
```

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

### Create New ListenMux

Source: https://github.com/storj/drpc/blob/main/drpcmigrate/README.md

NewListenMux initializes a ListenMux. It takes a base listener and a prefix length to determine how many initial bytes to inspect for routing decisions.

```go
func NewListenMux(base net.Listener, prefixLen int) *ListenMux
```

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

### Get Unblocked Channel for Manager in drpcmanager

Source: https://github.com/storj/drpc/blob/main/drpcmanager/README.md

Return a channel that closes when the manager is no longer blocked from creating a new stream due to a previous stream's soft cancel. This should not be called concurrently with stream creation methods.

```go
func (m *Manager) Unblocked() <-chan struct{}
```

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

### DRPC Stream Initialization

Source: https://context7.com/storj/drpc/llms.txt

Functions to create new DRPC streams, either with default or custom options.

```go
// New returns a new stream bound to the context with the given stream id.
func New(ctx context.Context, sid uint64, wr *drpcwire.Writer) *Stream

// NewWithOptions returns a new stream with custom options.
func NewWithOptions(ctx context.Context, sid uint64, wr *drpcwire.Writer, opts Options) *Stream
```

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

### Create New Tracker

Source: https://github.com/storj/drpc/blob/main/drpctest/README.md

NewTracker initializes a Tracker. It inspects the provided testing.TB to detect failures in launched goroutines.

```go
func NewTracker(tb testing.TB) *Tracker
```

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

### Get Connection from Pool

Source: https://github.com/storj/drpc/blob/main/drpcpool/README.md

Retrieves a connection for a given key. If no cached connection exists, it uses the provided dial function to create and cache a new one. It shares cached connections among multiple callers using the same key.

```go
func (p *Pool[K, V]) Get(ctx context.Context, key K, 
	dial func(ctx context.Context, key K) (V, error)) Conn
```

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

### Input Message String Method

Source: https://github.com/storj/drpc/blob/main/internal/integration/README.md

Returns a string representation of the In message.

```go
func (m *In) String() string
```

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

### Run gRPC and DRPC on Same Port in Go

Source: https://context7.com/storj/drpc/llms.txt

Use drpcmigrate.NewListenMux to route incoming traffic based on headers, allowing both gRPC and DRPC to share a single listener. Requires registering services with both gRPC and DRPC servers.

```go
package main

import (
    "context"
    "net"

    "golang.org/x/sync/errgroup"
    "google.golang.org/grpc"

    "storj.io/drpc/drpcmigrate"
    "storj.io/drpc/drpcmux"
    "storj.io/drpc/drpcserver"

    "example.com/myservice/pb"
)

type CookieMonsterServer struct {
    pb.UnimplementedCookieMonsterServer
}

func (s *CookieMonsterServer) EatCookie(ctx context.Context, cookie *pb.Cookie) (*pb.Crumbs, error) {
    return &pb.Crumbs{Cookie: cookie}, nil
}

func main() {
    cookieMonster := &CookieMonsterServer{}

    // Listen on a tcp socket
    lis, _ := net.Listen("tcp", ":8080")

    // Create a listen mux to route traffic based on header
    lisMux := drpcmigrate.NewListenMux(lis, len(drpcmigrate.DRPCHeader))
    drpcLis := lisMux.Route(drpcmigrate.DRPCHeader)
    grpcLis := lisMux.Default()

    var group errgroup.Group

    // gRPC handling (default route)
    group.Go(func() error {
        s := grpc.NewServer()
        pb.RegisterCookieMonsterServer(s, cookieMonster)
        return s.Serve(grpcLis)
    })

    // DRPC handling (DRPC header route)
    group.Go(func() error {
        m := drpcmux.New()
        pb.DRPCRegisterCookieMonster(m, cookieMonster)
        s := drpcserver.New(m)
        return s.Serve(context.Background(), drpcLis)
    })

    // Run the listen mux
    group.Go(func() error {
        return lisMux.Run(context.Background())
    })

    group.Wait()
}
```

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

### Create DRPC Service Client

Source: https://github.com/storj/drpc/blob/main/internal/integration/README.md

Constructs a new DRPC service client from an established DRPC connection. Use this to initiate communication with a DRPC service.

```go
func NewDRPCServiceClient(cc drpc.Conn) DRPCServiceClient
```

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

### Unimplemented DRPCServiceServer Method2

Source: https://github.com/storj/drpc/blob/main/internal/integration/README.md

Default implementation for Method2 on DRPCServiceUnimplementedServer. It should be overridden by concrete server implementations.

```go
func (s *DRPCServiceUnimplementedServer) Method2(DRPCService_Method2Stream) error
```

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

### DRPC Server Creation

Source: https://github.com/storj/drpc/blob/main/drpcserver/README.md

Provides methods for creating new DRPC server instances, either with default options or custom configurations.

```APIDOC
## func New

```go
func New(handler drpc.Handler) *Server
```
New constructs a new Server.

## func NewWithOptions

```go
func NewWithOptions(handler drpc.Handler, opts Options) *Server
```
NewWithOptions constructs a new Server using the provided options to tune how
the drpc connections are handled.
```

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

### Output Message String Method

Source: https://github.com/storj/drpc/blob/main/internal/integration/README.md

Returns a string representation of the Out message.

```go
func (m *Out) String() string
```

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

### Create a New Goroutine Tracker

Source: https://github.com/storj/drpc/blob/main/drpcctx/README.md

NewTracker creates a Tracker instance bound to the provided context. Use this to manage and track goroutines launched within the scope of this context.

```go
func NewTracker(ctx context.Context) *Tracker
```

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

### Kind String Representation

Source: https://github.com/storj/drpc/blob/main/drpcwire/README.md

Returns a human-readable string representation of a Kind.

```go
func (i Kind) String() string
```

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

### Server Implementation with drpcmux

Source: https://context7.com/storj/drpc/llms.txt

Details on using the `drpcmux` package for dispatching RPCs to registered service implementations on the server side.

```APIDOC
## Server Implementation

### drpcmux - RPC Multiplexer
Package `drpcmux` dispatches RPCs to registered service implementations.

```go
import "storj.io/drpc/drpcmux"

// New constructs a new Mux.
func New() *Mux

// Register associates the RPCs described by the description in the server.
func (m *Mux) Register(srv interface{}, desc drpc.Description) error

// HandleRPC handles the rpc that has been requested by the stream.
func (m *Mux) HandleRPC(stream drpc.Stream, rpc string) (err error)
```
```