Skip to content

Commit c4d097e

Browse files
hayabusa-cloudhayabusa-cloud
committed
feat: add portable message framing with non-blocking semantics
1 parent eded02c commit c4d097e

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

48 files changed

+6199
-2
lines changed

‎.github/workflows/ci.yml‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
name: CI
2+
3+
on:
4+
push:
5+
branches: [ main ]
6+
pull_request:
7+
branches: [ main ]
8+
9+
permissions:
10+
contents: read
11+
12+
concurrency:
13+
group: ci-${{ github.workflow }}-${{ github.ref }}
14+
cancel-in-progress: true
15+
16+
jobs:
17+
test:
18+
name: go test (${{ matrix.os }})
19+
runs-on: ${{ matrix.os }}
20+
strategy:
21+
matrix:
22+
os: [ubuntu-latest, windows-latest, macos-latest]
23+
go-version: [ '1.25.x' ]
24+
25+
steps:
26+
- name: Checkout
27+
uses: actions/checkout@v4
28+
29+
- name: Setup Go
30+
uses: actions/setup-go@v5
31+
with:
32+
go-version: ${{ matrix.go-version }}
33+
cache: true
34+
35+
- name: Go env
36+
run: go env
37+
38+
- name: Vet
39+
run: go vet ./...
40+
41+
- name: Test
42+
if: matrix.os != 'ubuntu-latest'
43+
run: go test ./... -count=1
44+
45+
- name: Test (with coverage)
46+
if: matrix.os == 'ubuntu-latest'
47+
run: go test ./... -count=1 -covermode=atomic -coverprofile=coverage.out
48+
49+
- name: Upload coverage to Codecov
50+
if: matrix.os == 'ubuntu-latest'
51+
uses: codecov/codecov-action@v5
52+
with:
53+
files: coverage.out
54+
flags: framer
55+
name: framer
56+
57+
# Runs tests on real runner CPU architectures that GitHub-hosted runners provide.
58+
native-arch:
59+
name: go test (native ${{ matrix.target.goos }}/${{ matrix.target.goarch }})
60+
runs-on: ${{ matrix.target.runsOn }}
61+
strategy:
62+
fail-fast: false
63+
matrix:
64+
target:
65+
- { runsOn: ubuntu-latest, goos: linux, goarch: amd64 }
66+
- { runsOn: windows-latest, goos: windows, goarch: amd64 }
67+
- { runsOn: macos-latest, goos: darwin, goarch: arm64 }
68+
69+
steps:
70+
- name: Checkout
71+
uses: actions/checkout@v4
72+
73+
- name: Setup Go
74+
uses: actions/setup-go@v5
75+
with:
76+
go-version: '1.25.x'
77+
cache: true
78+
79+
- name: Go env
80+
env:
81+
GOOS: ${{ matrix.target.goos }}
82+
GOARCH: ${{ matrix.target.goarch }}
83+
run: go env GOOS GOARCH
84+
85+
- name: Test (native)
86+
run: go test ./... -count=1
87+
88+
# Cross-compilation coverage for additional ports (compile only; no runners required).
89+
cross-compile:
90+
name: cross-compile (compile tests only)
91+
runs-on: ubuntu-latest
92+
strategy:
93+
fail-fast: false
94+
matrix:
95+
target:
96+
- { goos: linux, goarch: 386 }
97+
- { goos: linux, goarch: ppc64le }
98+
- { goos: linux, goarch: s390x }
99+
- { goos: linux, goarch: riscv64 }
100+
- { goos: linux, goarch: mips64le }
101+
- { goos: linux, goarch: mipsle }
102+
- { goos: linux, goarch: mips }
103+
104+
steps:
105+
- name: Checkout
106+
uses: actions/checkout@v4
107+
108+
- name: Setup Go
109+
uses: actions/setup-go@v5
110+
with:
111+
go-version: '1.25.x'
112+
cache: true
113+
114+
- name: Compile test binaries (${{ matrix.target.goos }}/${{ matrix.target.goarch }})
115+
shell: bash
116+
env:
117+
GOOS: ${{ matrix.target.goos }}
118+
GOARCH: ${{ matrix.target.goarch }}
119+
CGO_ENABLED: "0"
120+
run: |
121+
set -euo pipefail
122+
go env GOOS GOARCH CGO_ENABLED
123+
124+
outdir="$(mktemp -d)"
125+
ext=""
126+
if [ "${GOOS}" = "windows" ]; then ext=".exe"; fi
127+
128+
# Build (compile only) each package's test binary for the target platform.
129+
while IFS= read -r pkg; do
130+
name="$(echo "${pkg}" | tr '/.' '__')"
131+
go test -c "${pkg}" -o "${outdir}/${name}${ext}"
132+
done < <(go list ./...)

‎README.es.md‎

Lines changed: 214 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,214 @@
1+
# framer — límites de mensaje sobre E/S de flujo
2+
3+
[![Go Reference](https://pkg.go.dev/badge/code.hybscloud.com/framer.svg)](https://pkg.go.dev/code.hybscloud.com/framer)
4+
[![Go Report Card](https://goreportcard.com/badge/github.com/hayabusa-cloud/framer)](https://goreportcard.com/report/github.com/hayabusa-cloud/framer)
5+
[![Coverage Status](https://codecov.io/gh/hayabusa-cloud/framer/graph/badge.svg)](https://codecov.io/gh/hayabusa-cloud/framer)
6+
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)
7+
8+
**Idiomas / Languages:** [English](README.md) | [简体中文](README.zh-CN.md) | [日本語](README.ja.md) | Español | [Français](README.fr.md)
9+
10+
Framing de mensajes portable para Go. Conserva “un mensaje por `Read`/`Write`” sobre transportes tipo stream.
11+
12+
Alcance: `framer` resuelve la preservación de límites de mensaje en transportes de flujo.
13+
14+
## En resumen
15+
16+
- Resuelve problemas de límites de mensaje en flujos de bytes (TCP, Unix stream, pipes).
17+
- Pass-through en transportes que ya preservan límites (UDP, Unix datagram, WebSocket, SCTP).
18+
- Formato wire portable; orden de bytes configurable.
19+
20+
## Por qué
21+
22+
Muchos transportes son flujos de bytes (TCP, Unix stream, pipes). Un solo `Read` puede devolver una fracción de un mensaje de aplicación, o varios mensajes concatenados. `framer` restaura los límites: en modo stream, un `Read` devuelve exactamente un payload de mensaje y un `Write` emite exactamente un mensaje enmarcado.
23+
24+
## Adaptación de protocolo
25+
26+
- `BinaryStream` (transportes stream: TCP, TLS-over-TCP, Unix stream, pipes): agrega un prefijo de longitud; lee/escribe mensajes completos.
27+
- `SeqPacket` (p. ej., SCTP, WebSocket): pass-through; el transporte ya preserva límites.
28+
- `Datagram` (p. ej., UDP, Unix datagram): pass-through; el transporte ya preserva límites.
29+
30+
Selecciona al construir vía `WithProtocol(...)` (hay variantes de lectura/escritura) o usa los helpers de transporte (ver Options).
31+
32+
## Wire format
33+
34+
Prefijo de longitud compacto de tamaño variable, seguido por bytes de payload. El orden de bytes para la longitud extendida es configurable: `WithByteOrder`, o por dirección `WithReadByteOrder` / `WithWriteByteOrder`.
35+
36+
## Formato de datos del frame
37+
38+
El esquema de framing de `framer` es intencionalmente compacto:
39+
40+
- Byte de cabecera `H0` + bytes opcionales de longitud extendida.
41+
- Sea `L` la longitud del payload en bytes.
42+
- Si `0 ≤ L ≤ 253` (`0x00..0xFD`): `H0 = L`. Sin bytes extra.
43+
- Si `254 ≤ L ≤ 65535` (`0x0000..0xFFFF`): `H0 = 0xFE` y los siguientes 2 bytes codifican `L` como entero sin signo de 16 bits en el orden configurado.
44+
- Si `65536 ≤ L ≤ 2^56-1`: `H0 = 0xFF` y los siguientes 7 bytes llevan los 56 bits bajos de `L` en el orden configurado.
45+
- Big-endian: bytes `[1..7]` son los 56 bits bajos de `L` en big-endian.
46+
- Little-endian: bytes `[1..7]` son los 56 bits bajos de `L` en little-endian.
47+
48+
Límites y errores:
49+
- La longitud máxima de payload soportada es `2^56-1`; valores mayores producen `framer.ErrTooLong`.
50+
- Con un límite de lectura (`WithReadLimit`), longitudes mayores fallan con `framer.ErrTooLong`.
51+
52+
## Inicio rápido
53+
54+
Instala con `go get`:
55+
```shell
56+
go get code.hybscloud.com/framer
57+
```
58+
59+
```go
60+
c1, c2 := net.Pipe()
61+
defer c1.Close()
62+
defer c2.Close()
63+
64+
w := framer.NewWriter(c1, framer.WithWriteTCP())
65+
r := framer.NewReader(c2, framer.WithReadTCP())
66+
67+
go func() { _, _ = w.Write([]byte("hello")) }()
68+
69+
buf := make([]byte, 64)
70+
n, err := r.Read(buf)
71+
if err != nil {
72+
panic(err)
73+
}
74+
fmt.Printf("got: %q\n", buf[:n])
75+
```
76+
77+
## Options
78+
79+
- `WithProtocol(proto Protocol)` — elige `BinaryStream`, `SeqPacket` o `Datagram` (hay variantes de lectura/escritura).
80+
- Orden de bytes: `WithByteOrder`, o `WithReadByteOrder` / `WithWriteByteOrder`.
81+
- `WithReadLimit(n int)` — limita el tamaño máximo del payload al leer.
82+
- `WithRetryDelay(d time.Duration)` — política de would-block; helpers: `WithNonblock()` / `WithBlock()`.
83+
84+
Helpers de transporte (presets):
85+
- `WithReadTCP` / `WithWriteTCP` (`BinaryStream`, BigEndian en orden de red)
86+
- `WithReadUDP` / `WithWriteUDP` (`Datagram`, BigEndian)
87+
- `WithReadWebSocket` / `WithWriteWebSocket` (`SeqPacket`, BigEndian)
88+
- `WithReadSCTP` / `WithWriteSCTP` (`SeqPacket`, BigEndian)
89+
- `WithReadUnix` / `WithWriteUnix` (`BinaryStream`, BigEndian)
90+
- `WithReadUnixPacket` / `WithWriteUnixPacket` (`Datagram`, BigEndian)
91+
- `WithReadLocal` / `WithWriteLocal` (`BinaryStream`, orden nativo)
92+
93+
Más: GoDoc https://pkg.go.dev/code.hybscloud.com/framer
94+
95+
## Contrato de semántica (Semantics Contract)
96+
97+
### Taxonomía de errores
98+
99+
| Error | Significado | Acción del llamador |
100+
|-------|-------------|---------------------|
101+
| `nil` | Operación completada con éxito | Continúa; `n` refleja el progreso total |
102+
| `io.EOF` | Fin de stream (no hay más mensajes) | Deja de leer; terminación normal |
103+
| `io.ErrUnexpectedEOF` | El stream terminó a mitad de mensaje (header o payload incompleto) | Trátalo como fatal; posible corrupción o desconexión |
104+
| `io.ErrShortBuffer` | Buffer destino demasiado pequeño para el payload | Reintenta con un buffer más grande |
105+
| `io.ErrShortWrite` | El destino aceptó menos bytes que los provistos | Reintenta o trátalo como fatal según el contexto |
106+
| `io.ErrNoProgress` | El Reader subyacente no avanzó (`n==0, err==nil`) con buffer no vacío | Trátalo como fatal; indica un `io.Reader` roto |
107+
| `framer.ErrWouldBlock` | No es posible avanzar ahora sin esperar | Reintenta más tarde (tras poll/event); `n` puede ser >0 |
108+
| `framer.ErrMore` | Hubo progreso; seguirán más completions del mismo op | Procesa y vuelve a llamar |
109+
| `framer.ErrTooLong` | El mensaje excede límites o el máximo del wire format | Rechaza; posiblemente fatal |
110+
| `framer.ErrInvalidArgument` | Reader/Writer nil o configuración inválida | Corrige la configuración |
111+
112+
### Tablas de resultados
113+
114+
**`Reader.Read(p []byte) (n int, err error)`** — modo BinaryStream
115+
116+
| Condición | n | err |
117+
|----------|---|-----|
118+
| Mensaje completo entregado | payload length | `nil` |
119+
| `len(p) < payload length` | 0 | `io.ErrShortBuffer` |
120+
| Payload excede ReadLimit | 0 | `ErrTooLong` |
121+
| El subyacente devuelve would-block | bytes leídos hasta ahora | `ErrWouldBlock` |
122+
| El subyacente devuelve more | bytes leídos hasta ahora | `ErrMore` |
123+
| EOF en el límite de mensaje | 0 | `io.EOF` |
124+
| EOF a mitad de header o payload | bytes leídos | `io.ErrUnexpectedEOF` |
125+
126+
**`Writer.Write(p []byte) (n int, err error)`** — modo BinaryStream
127+
128+
| Condición | n | err |
129+
|----------|---|-----|
130+
| Mensaje enmarcado completo emitido | `len(p)` | `nil` |
131+
| Payload excede el máximo (2^56-1) | 0 | `ErrTooLong` |
132+
| El subyacente devuelve would-block | bytes de payload escritos | `ErrWouldBlock` |
133+
| El subyacente devuelve more | bytes de payload escritos | `ErrMore` |
134+
135+
**`Reader.WriteTo(dst io.Writer) (n int64, err error)`**
136+
137+
| Condición | n | err |
138+
|----------|---|-----|
139+
| Transferencia hasta EOF | bytes totales de payload | `nil` |
140+
| Reader subyacente devuelve would-block | bytes de payload escritos | `ErrWouldBlock` |
141+
| Reader subyacente devuelve more | bytes de payload escritos | `ErrMore` |
142+
| `dst` devuelve would-block | bytes de payload escritos | `ErrWouldBlock` |
143+
| Mensaje excede el buffer interno (64KiB por defecto) | bytes hasta ahora | `ErrTooLong` |
144+
| Stream terminó a mitad de mensaje | bytes hasta ahora | `io.ErrUnexpectedEOF` |
145+
146+
**`Writer.ReadFrom(src io.Reader) (n int64, err error)`**
147+
148+
| Condición | n | err |
149+
|----------|---|-----|
150+
| Chunks codificados hasta src EOF | bytes totales de payload | `nil` |
151+
| `src` devuelve would-block | bytes de payload escritos | `ErrWouldBlock` |
152+
| `src` devuelve more | bytes de payload escritos | `ErrMore` |
153+
| Writer subyacente devuelve would-block | bytes de payload escritos | `ErrWouldBlock` |
154+
155+
**`Forwarder.ForwardOnce() (n int, err error)`**
156+
157+
| Condición | n | err |
158+
|----------|---|-----|
159+
| Un mensaje reenviado completamente | bytes de payload (fase de escritura) | `nil` |
160+
| Fuente packet devuelve `(n>0, io.EOF)` | bytes de payload (fase de escritura) | `nil` (la próxima llamada devuelve `io.EOF`) |
161+
| No hay más mensajes | 0 | `io.EOF` |
162+
| Would-block en fase de lectura | bytes leídos en esta llamada | `ErrWouldBlock` |
163+
| Would-block en fase de escritura | bytes escritos en esta llamada | `ErrWouldBlock` |
164+
| Mensaje excede el buffer interno | 0 | `io.ErrShortBuffer` |
165+
| Mensaje excede ReadLimit | 0 | `ErrTooLong` |
166+
| Stream terminó a mitad de mensaje | bytes hasta ahora | `io.ErrUnexpectedEOF` |
167+
168+
### Clasificación de operaciones
169+
170+
| Operación | Comportamiento de límites | Caso de uso |
171+
|----------|----------------------------|------------|
172+
| `Reader.Read` | **Preserva límites**: 1 llamada = 1 mensaje | Procesamiento por mensaje |
173+
| `Writer.Write` | **Preserva límites**: 1 llamada = 1 mensaje | Envío por mensaje |
174+
| `Reader.WriteTo` | **Chunking**: stream de bytes de payload (no wire format) | Transferencia eficiente; NO preserva límites |
175+
| `Writer.ReadFrom` | **Chunking**: cada chunk de `src` se vuelve un mensaje | Codificación eficiente; NO preserva límites aguas arriba |
176+
| `Forwarder.ForwardOnce` | **Relay con límites**: decodifica uno, re-encodifica uno | Proxy con preservación de límites |
177+
178+
### Política de bloqueo
179+
180+
Por defecto, framer es **no bloqueante** (`WithNonblock()`): devuelve `ErrWouldBlock` inmediatamente.
181+
182+
- `WithBlock()` — hace yield (`runtime.Gosched`) y reintenta ante would-block
183+
- `WithRetryDelay(d)` — duerme `d` y reintenta ante would-block
184+
- `RetryDelay` negativo (por defecto) — devuelve `ErrWouldBlock` inmediatamente
185+
186+
Ningún método oculta bloqueo a menos que se configure explícitamente.
187+
188+
## Fast paths
189+
190+
`framer` implementa fast paths del stdlib para interoperar con motores tipo `io.Copy` y con `iox.CopyPolicy`:
191+
192+
- `(*Reader).WriteTo(io.Writer)` — transfiere eficientemente payloads a `dst`.
193+
- Stream (`BinaryStream`): procesa un mensaje por vez y escribe solo bytes de payload. Si `ReadLimit == 0`, usa un tope conservador (64KiB); mensajes más grandes devuelven `framer.ErrTooLong`.
194+
- Packet (`SeqPacket`/`Datagram`): pass-through.
195+
- `framer.ErrWouldBlock` y `framer.ErrMore` se propagan sin cambios, con el conteo reflejando bytes escritos.
196+
197+
- `(*Writer).ReadFrom(io.Reader)` — chunk-to-message: cada chunk leído de `src` se codifica como un mensaje y se escribe vía `w.Write`.
198+
- Es eficiente pero no preserva límites de mensaje de `src`.
199+
- En protocolos con límites preservados, se comporta como pass-through.
200+
- `framer.ErrWouldBlock` y `framer.ErrMore` se propagan sin cambios.
201+
202+
Recomendación: en bucles no bloqueantes, prefiere `iox.CopyPolicy` con política de reintentos (p. ej., `PolicyRetry`) para manejar explícitamente `ErrWouldBlock` / `ErrMore`.
203+
204+
## Reenvío
205+
206+
- Proxy a nivel wire (motores de bytes): usa `iox.CopyPolicy` y fast paths estándar (`WriterTo`/`ReaderFrom`). Maximiza throughput cuando no necesitas preservar límites de nivel superior.
207+
- Relay por mensaje (preserva límites): usa `framer.NewForwarder(dst, src, ...)` y llama `ForwardOnce()` en tu poll loop. Decodifica exactamente un mensaje desde `src` y lo re-encodifica como exactamente un mensaje hacia `dst`.
208+
- Semántica no bloqueante: `ForwardOnce` devuelve `(n>0, framer.ErrWouldBlock|framer.ErrMore)` cuando hubo progreso parcial; reintenta con la misma instancia.
209+
- Límites: `io.ErrShortBuffer` si el buffer interno es insuficiente; `framer.ErrTooLong` si excede `WithReadLimit`.
210+
- Cero asignaciones en steady-state tras la construcción; el buffer interno se reutiliza.
211+
212+
## Licencia
213+
214+
MIT — ver `LICENSE`.

0 commit comments

Comments
 (0)