Cancel a stream in a streaming client.

After that, callings to recv/2 will return a CANCEL error.

@spec connect(
  String.t(),
  keyword()
) :: {:ok, GRPC.Channel.t()} | {:error, any()}

Establishes a connection with a gRPC server and returns a GRPC.Channel required for sending requests. Supports advanced connection resolution via the gRPC Resolver and various target schemes (dns, unix, xds, host:port, etc).

This function is part of the connection orchestration layer, which manages connection setup, name resolution, and optional load balancing.

Target Syntax

The target argument to connect/2 accepts URI-like strings that are resolved using the configured Resolver.

Supported formats:

• "dns://example.com:50051" — resolves via DNS (A/AAAA records and _grpc_config TXT)
• "ipv4:10.0.0.5:50051" — fixed IPv4 address
• "unix:/tmp/my.sock" — Unix domain socket
• "xds:///my-service" — resolves via xDS control plane (Envoy/Istio/Traffic Director)
• "127.0.0.1:50051" — implicit DNS (default port 50051)

If no scheme is provided, the resolver assumes dns by default.

Options

• :cred - a GRPC.Credential used to indicate it's a secure connection. An insecure connection will be created without this option.
• :adapter - custom client adapter
• :interceptors - client interceptors
• :codec - client will use this to encode and decode binary message
• :compressor - the client will use this to compress requests and decompress responses. If this is set, accepted_compressors will be appended also, so this can be used safely without :accepted_compressors.
• :accepted_compressors - tell servers accepted compressors, this can be used without :compressor
• :headers - headers to attach to each request

Examples

Basic Connection

iex> GRPC.Stub.connect("localhost:50051")
{:ok, channel}

iex> GRPC.Stub.connect("localhost:50051", accepted_compressors: [GRPC.Compressor.Gzip])
{:ok, channel}

DNS Target

iex> {:ok, ch} = GRPC.Client.Connection.connect("dns://my-service.local:50051")

Unix Socket

iex> GRPC.Stub.connect("/path/to/unix.sock")
{:ok, channel}

Notes

• When using DNS or xDS targets, the connection layer periodically refreshes endpoints.

@spec disconnect(GRPC.Channel.t()) :: {:ok, GRPC.Channel.t()} | {:error, any()}

Disconnects the adapter and frees any resources the adapter is consuming

@spec end_stream(GRPC.Client.Stream.t()) :: GRPC.Client.Stream.t()

Send END_STREAM frame to end the stream.

The stream will be in half_closed state after this is called.

Examples

iex> stream = GRPC.Stub.send_request(stream, request)
iex> GRPC.Stub.end_stream(stream)

@spec recv(
  GRPC.Client.Stream.t(),
  keyword()
) ::
  {:ok, struct()}
  | {:ok, struct(), map()}
  | {:ok, Enumerable.t()}
  | {:ok, Enumerable.t(), map()}
  | {:error, any()}

Receive replies when requests are streaming.

• If the reply is not streaming, a normal reply struct will be returned
• If the reply is streaming, a enumerable Stream will be returned. You can use Enum to fetch further replies or Stream to manipulate it. Each item in the Enumerable is a tuple {:ok, reply} or {:error, error}. When :return_headers is true, the last item in the Enumerable will be {:trailers, map}

Examples

# Reply is not streaming
{:ok, reply} = GRPC.Stub.recv(stream)

# Reply is streaming
{:ok, ex_stream} = GRPC.Stub.recv(stream)
replies = Enum.map(ex_stream, fn({:ok, reply}) -> reply end)

Options

• :timeout - request timeout
• :deadline - when the request is timeout, will override timeout
• :return_headers - when true, headers will be returned.

Stream behavior

We build the Stream struct using Stream.unfold/2.

The unfold function is built in such a way that - for both adapters - the accumulator is a map used to find the connection_streamprocess and the next_fun argument is a function that reads directly from the connection_stream that is producing data. Every time we execute next_fun we read a chunk of data. This means that next_fun will have the side effect of updating the state of the connection_stream process, removing the chunk of data that's being read from the underlying GenServer's state.

Examples

iex> ex_stream |> Stream.take(1) |> Enum.to_list()
[1]
iex> ex_stream |> Enum.to_list()
[2, 3]
iex> ex_stream |> Enum.to_list()
[]

@spec send_request(GRPC.Client.Stream.t(), struct(), keyword()) ::
  GRPC.Client.Stream.t()

Send streaming requests.

The last request can be sent with :end_stream option, or you can call end_stream/1 to send a frame with END_STREAM flag to end the stream.

Options

• :end_stream - indicates it's the last one request, then the stream will be in half_closed state. Default is false.