NodeDatagramSocket

Node UDP adapter for effect/socket/DatagramSocket, built on node:dgram.

make opens a new dgram.Socket for each reader acquisition, fromSocket adopts a socket the caller creates, and layer provides the DatagramSocket service.

The reader owns the native socket, so a send-only client must still acquire a reader, or its writes wait forever. Every write waits for Node's send callback, so send errors fail the write that caused them.

writeAll fails with the first send error. Which of the other datagrams went out is unspecified and nothing is resent; the error carries the failed datagram's address.

Example Ingest server

import { NodeDatagramSocket } from "@effect/platform-node-shared"
import { Effect } from "effect"
import { DatagramSocket } from "effect/socket"

const ingest = Effect.gen(function*() {
  const socket = yield* DatagramSocket.DatagramSocket
  const reader = yield* socket.reader
  while (true) {
    const datagrams = yield* reader.pull
    yield* Effect.log(`received ${datagrams.length}, dropped ${reader.dropped()}`)
  }
}).pipe(
  Effect.scoped,
  Effect.provide(NodeDatagramSocket.layer({
    bind: { port: 9000 },
    receiveBuffer: { capacity: 4096, strategy: "sliding" }
  }))
)

Example Echo server

import { NodeDatagramSocket } from "@effect/platform-node-shared"
import { Effect } from "effect"

const echo = Effect.gen(function*() {
  const socket = yield* NodeDatagramSocket.make({ bind: { port: 9000 } })
  const reader = yield* socket.reader
  const writer = yield* socket.writer
  while (true) {
    for (const received of yield* reader.pull) {
      yield* writer.write({ payload: received.payload, address: received })
    }
  }
}).pipe(Effect.scoped)

Example Send-only client with a connected socket

import { NodeDatagramSocket } from "@effect/platform-node-shared"
import { Effect } from "effect"

const send = (payload: string) =>
  Effect.gen(function*() {
    const socket = yield* NodeDatagramSocket.make({
      connect: { address: "localhost", port: 9000 }
    })
    // Required even for send-only clients.
    yield* socket.reader
    const writer = yield* socket.writer
    yield* writer.write({ payload })
  }).pipe(Effect.scoped)

Example Multicast receiver and sender

import { NodeDatagramSocket } from "@effect/platform-node-shared"
import { Effect } from "effect"
import { NetAddress } from "effect/net"

const group = NetAddress.ipFromStringUnsafe("239.255.0.1")

const receive = Effect.gen(function*() {
  if (!NetAddress.isMulticast(group) || !NetAddress.isIpv4Address(group)) return
  const socket = yield* NodeDatagramSocket.make({
    bind: { address: "0.0.0.0", port: 5000 },
    reuseAddress: true
  })
  const reader = yield* socket.reader
  yield* reader.joinMulticast({ group, interface: NetAddress.ipv4Loopback })
  return yield* reader.pull
}).pipe(Effect.scoped)

const announce = Effect.gen(function*() {
  const socket = yield* NodeDatagramSocket.make({
    peer: { address: group, port: 5000 },
    // Egress interface, independent of the join's ingress interface.
    multicast: { interface: NetAddress.ipv4Loopback, ttl: 1, loopback: true }
  })
  yield* socket.reader
  const writer = yield* socket.writer
  yield* writer.write({ payload: "hello" })
}).pipe(Effect.scoped)

7 exports Since v4.0.0 Source

Constructors

fromSocket

Since v4.0.0

Adopts a dgram.Socket.

Details

acquire runs once per reader acquisition, inside the reader's scope, and the adapter closes the socket when that scope ends. The socket must be bound by the time acquire completes. Whether it is connected is checked once at open.

Signature

declare function fromSocket<R>(acquire: Effect<Socket, DatagramSocketError, R>, options: FromSocketOptions): Effect<DatagramSocket, never, Exclude<R, Scope>>

make

Since v4.0.0

Creates a DatagramSocket that opens and binds a new dgram.Socket for each reader acquisition.

Details

Creating the socket never fails. Binding, applying options and resolving names happen when a reader is acquired, and fail that acquisition with a DatagramSocketError. Hostnames in bind, peer and connect are resolved once per acquisition, so the native socket only sees IP literals and no send waits for DNS.

Invalid receiveBuffer.capacity values cause a defect; use a positive safe integer.

Signature

declare function make(options: Options): Effect<DatagramSocket>

Layers

layer

Since v4.0.0

Provides a DatagramSocket built with make.

Signature

declare function layer(options?: Options): Layer<DatagramSocket>

Models

BindOptions

interface

Since v4.0.0

The local address and port to bind. Defaults to 0.0.0.0 (or :: for family: "ipv6") and an ephemeral port.

Signature

interface BindOptions {
  readonly address?: string | IpAddress;
  readonly port?: number;
}

Endpoint

interface

Since v4.0.0

An open-time endpoint. Hostnames resolve once per reader acquisition via node:dns.lookup. An InetAddress can be passed directly.

Signature

interface Endpoint {
  readonly address: string | IpAddress;
  readonly port: number;
}

FromSocketOptions

interface

Since v4.0.0

Options for fromSocket. The caller configures the adopted socket, so only the options that live in JavaScript apply.

Signature

interface FromSocketOptions {
  readonly onError?: (error: DatagramSocketError) => void;
  readonly peer?: Endpoint;
  readonly receiveBuffer?: any;
}

Options

type

Since v4.0.0

Options for make and layer.

Details

peer is a default destination and doesn't filter senders. connect calls connect(2) after binding, so the kernel filters senders and ICMP errors reach onError. The two can't be combined.

Family selection, in order: explicit family, a bind IP literal, the peer or connect address, a bind hostname, then "ipv4". Hostname lookups use the family selected so far, or prefer IPv4 if none is set.

reuseAddress means SO_REUSEADDR on Linux and SO_REUSEPORT on BSD and macOS. kernelReceiveBufferSize and kernelSendBufferSize are in bytes; Linux doubles the value and caps it at rmem_max / wmem_max.

multicast.interface is the egress interface for datagrams sent to a group, which is a different socket option from the ingress interface of Reader.joinMulticast.

Signature

type Options = {
  readonly bind?: BindOptions;
  readonly broadcast?: boolean;
  readonly family?: "ipv4" | "ipv6";
  readonly ipv6Only?: boolean;
  readonly kernelReceiveBufferSize?: number;
  readonly kernelSendBufferSize?: number;
  readonly multicast?: {
    readonly interface?: NetAddress.MulticastInterface;
    readonly loopback?: boolean;
    readonly ttl?: number;
  };
  readonly onError?: (error: DatagramSocket.DatagramSocketError) => void;
  readonly receiveBuffer?: DatagramSocket.ReceiveBufferOptions;
  readonly reuseAddress?: boolean;
  readonly reusePort?: boolean;
  readonly ttl?: number;
} & {
  readonly connect?: undefined;
  readonly peer?: Endpoint;
} | {
  readonly connect: Endpoint;
  readonly peer?: undefined;
}