feat: init basic server

Author: aminofox

README.md

@@ -0,0 +1,100 @@
+# zsocket
+
+A small, gorilla-free WebSocket server for Go — clean API, RFC6455 handshake, frames, ping/pong keepalive, JSON helpers, and a simple room hub. Built to be extended (rate limiting, metrics, distributed hub) without leaking third-party types into your code.
+
+> Module: `github.com/aminofox/zsocket`
+
+## Features
+
+- Minimal RFC6455 server-side implementation
+- Clean API:
+  - `Accept(w, r, cfg) (*Conn, error)`
+  - `Conn.Read(ctx) (MessageType, []byte, error)`
+  - `Conn.Write(ctx, mt, data)`
+  - `Conn.ReadJSON/WriteJSON`
+  - `Conn.Close(code, reason)`
+  - `Conn.Subprotocol()`, `Conn.RemoteAddr()`
+- Fragmentation (basic accumulation until `FIN`)
+- Control frames: ping/pong/close
+- Keepalive: `PingInterval` + `PongWait`
+- `Hub`: rooms & broadcast
+- Configurable read limit, deadlines, origin policy, subprotocols
+
+## Quick Start
+
+```bash
+go get github.com/aminofox/zsocket
+```
+
+```code
+package main
+
+import (
+  "context"
+  "log"
+  "net/http"
+  "github.com/aminofox/zsocket"
+)
+
+func main() {
+  cfg := zsocket.DefaultConfig()
+  hub := zsocket.NewHub()
+
+  http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
+    conn, err := zsocket.Accept(w, r, cfg)
+    if err != nil {
+      http.Error(w, "upgrade failed", http.StatusBadRequest)
+      return
+    }
+    hub.Join("lobby", conn)
+
+    go func() {
+      defer func() {
+        hub.LeaveAll(conn)
+        _ = conn.Close(zsocket.CloseNormalClosure, "bye")
+      }()
+      ctx := context.Background()
+      _ = conn.Write(ctx, zsocket.TextMessage, []byte("hello from zsocket"))
+
+      for {
+        mt, msg, err := conn.Read(ctx)
+        if err != nil {
+          log.Printf("read: %v", err)
+          return
+        }
+        _ = conn.Write(ctx, mt, msg) // echo
+        hub.Broadcast(ctx, "lobby", mt, msg)
+      }
+    }()
+  })
+
+  log.Fatal(http.ListenAndServe(":8080", nil))
+}
+```
+
+## Config
+```
+cfg := zsocket.DefaultConfig()
+cfg.AllowedOrigins = []string{"https://example.com"} // or nil to allow all
+cfg.Subprotocols = []string{"chat", "json"}
+cfg.ReadLimit = 16 << 20  // 16 MiB per message
+cfg.PingInterval = 30 * time.Second
+cfg.PongWait = 15 * time.Second
+cfg.WriteTimeout = 10 * time.Second
+```
+
+## JSON Helpers
+```
+type In struct { Cmd string `json:"cmd"`; Body any `json:"body"` }
+
+var in In
+if err := conn.ReadJSON(ctx, &in); err != nil { ... }
+if err := conn.WriteJSON(ctx, map[string]any{"ok": true}); err != nil { ... }
+```
+
+## Rooms & Broadcast
+```
+hub.Join("room-42", conn)
+hub.Broadcast(ctx, "room-42", zsocket.TextMessage, []byte("hi all"))
+hub.Leave("room-42", conn)
+```

conn.go

@@ -0,0 +1,272 @@
+package zsocket
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Conn is a server-side WebSocket connection managed by zsocket.
+// It exposes Read/Write methods, JSON helpers, and graceful Close.
+// Concurrency: one reader goroutine internally; writes are serialized via writeMu.
+//
+// NOTE: This MVP supports basic fragmentation (accumulate until FIN). Extensions are not implemented.
+type Conn struct {
+	raw        net.Conn
+	br         *bufio.Reader
+	bw         *bufio.Writer
+	cfg        Config
+	subproto   string
+	remoteAddr string
+
+	readLimit int64
+
+	// pong keeper
+	lastPong atomic.Value // time.Time
+	closed   atomic.Bool
+
+	// Ensure only one write at a time.
+	writeMu sync.Mutex
+	// Allow readers to stop.
+	closeOnce sync.Once
+}
+
+// newConn wraps a hijacked net.Conn into a zsocket Conn with buffers and starts keepalive if configured.
+func newConn(c net.Conn, cfg Config, subproto, remote string) (*Conn, error) {
+	if cfg.ReadBufferSize <= 0 {
+		cfg.ReadBufferSize = DefaultConfig().ReadBufferSize
+	}
+	if cfg.WriteBufferSize <= 0 {
+		cfg.WriteBufferSize = DefaultConfig().WriteBufferSize
+	}
+	if cfg.ReadLimit < 0 {
+		cfg.ReadLimit = 0
+	}
+
+	z := &Conn{
+		raw:        c,
+		br:         bufio.NewReaderSize(c, cfg.ReadBufferSize),
+		bw:         bufio.NewWriterSize(c, cfg.WriteBufferSize),
+		cfg:        cfg,
+		subproto:   subproto,
+		remoteAddr: remote,
+		readLimit:  cfg.ReadLimit,
+	}
+	z.lastPong.Store(time.Now())
+
+	// Start keepalive loop if enabled.
+	if cfg.PingInterval > 0 && cfg.PongWait > 0 {
+		go z.keepAlive()
+	}
+
+	return z, nil
+}
+
+// keepAlive periodically sends ping frames and checks for timely pongs.
+// If a pong is not received within PongWait, the connection is closed.
+func (c *Conn) keepAlive() {
+	ticker := time.NewTicker(c.cfg.PingInterval)
+	defer ticker.Stop()
+
+	for {
+		if c.closed.Load() {
+			return
+		}
+		select {
+		case <-ticker.C:
+			_ = c.WriteControl(opPing, nil) // best-effort; ignore error
+			// check pong freshness
+			last := c.lastPong.Load().(time.Time)
+			if time.Since(last) > c.cfg.PongWait {
+				_ = c.Close(CloseGoingAway, "pong timeout")
+				return
+			}
+		}
+	}
+}
+
+// Subprotocol returns the negotiated subprotocol, if any.
+func (c *Conn) Subprotocol() string { return c.subproto }
+
+// RemoteAddr returns the peer address as string.
+func (c *Conn) RemoteAddr() string { return c.remoteAddr }
+
+// Close sends a close frame (best-effort) and closes the connection.
+func (c *Conn) Close(code CloseCode, reason string) error {
+	var err error
+	c.closeOnce.Do(func() {
+		c.closed.Store(true)
+		// Send close control frame (best-effort).
+		_ = c.WriteControl(opClose, closePayload(code, reason))
+		err = c.raw.Close()
+	})
+	return err
+}
+
+// Read reads the next complete message, returning its type and payload.
+// It accumulates fragments until FIN=1. Control frames are handled internally.
+func (c *Conn) Read(ctx context.Context) (MessageType, []byte, error) {
+	// Context deadline/timeout support on the underlying conn.
+	if deadline, ok := ctx.Deadline(); ok {
+		_ = c.raw.SetReadDeadline(deadline)
+		defer c.raw.SetReadDeadline(time.Time{})
+	}
+
+	var (
+		msgOpcode byte
+		full      bytes.Buffer
+		started   bool
+	)
+
+	for {
+		fin, opcode, payload, err := readFrame(c.br, c.readLimit)
+		if err != nil {
+			return 0, nil, err
+		}
+
+		switch opcode {
+		case opText, opBinary:
+			if started {
+				return 0, nil, errFragmentState
+			}
+			msgOpcode = opcode
+			full.Write(payload)
+			started = true
+			if fin {
+				return toMsgType(msgOpcode), full.Bytes(), nil
+			}
+
+		case opContinuation:
+			if !started {
+				return 0, nil, errUnexpectedOpcode
+			}
+			full.Write(payload)
+			if fin {
+				return toMsgType(msgOpcode), full.Bytes(), nil
+			}
+
+		case opPing:
+			// Immediately respond with pong (mirror payload).
+			_ = c.WriteControl(opPong, payload)
+
+		case opPong:
+			// Update lastPong marker.
+			c.lastPong.Store(time.Now())
+
+		case opClose:
+			code, reason := parseClosePayload(payload)
+			// Echo close (best-effort) and close connection locally.
+			_ = c.WriteControl(opClose, closePayload(code, reason))
+			return 0, nil, CloseError{Code: code, Reason: reason}
+
+		default:
+			// Protocol error → close
+			_ = c.Close(CloseProtocolError, "unsupported opcode")
+			return 0, nil, fmt.Errorf("zsocket: unsupported opcode %d", opcode)
+		}
+	}
+}
+
+// Write sends a complete message (text or binary) as a single unfragmented frame.
+func (c *Conn) Write(ctx context.Context, mt MessageType, data []byte) error {
+	if c.closed.Load() {
+		return ErrClosed
+	}
+	if mt != TextMessage && mt != BinaryMessage {
+		return fmt.Errorf("zsocket: invalid message type %d", mt)
+	}
+	// Apply write deadline if ctx has one or from config.
+	if deadline, ok := ctx.Deadline(); ok {
+		_ = c.raw.SetWriteDeadline(deadline)
+		defer c.raw.SetWriteDeadline(time.Time{})
+	} else if c.cfg.WriteTimeout > 0 {
+		_ = c.raw.SetWriteDeadline(time.Now().Add(c.cfg.WriteTimeout))
+		defer c.raw.SetWriteDeadline(time.Time{})
+	}
+
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
+
+	var opcode byte
+	if mt == TextMessage {
+		opcode = opText
+	} else {
+		opcode = opBinary
+	}
+
+	if err := writeFrame(c.bw, true, opcode, data); err != nil {
+		return err
+	}
+	return c.bw.Flush()
+}
+
+// WriteJSON marshals v to JSON and writes as a text message.
+func (c *Conn) WriteJSON(ctx context.Context, v any) error {
+	b, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	return c.Write(ctx, TextMessage, b)
+}
+
+// ReadJSON reads next message and unmarshals JSON into v.
+// It accepts both text and binary frames. If binary, it still tries to decode JSON.
+func (c *Conn) ReadJSON(ctx context.Context, v any) error {
+	mt, b, err := c.Read(ctx)
+	if err != nil {
+		return err
+	}
+	if mt != TextMessage && mt != BinaryMessage {
+		return fmt.Errorf("zsocket: unexpected message type %d", mt)
+	}
+	return json.Unmarshal(b, v)
+}
+
+// WriteControl writes a control frame (ping/pong/close). Best-effort utility.
+func (c *Conn) WriteControl(opcode byte, payload []byte) error {
+	if c.closed.Load() {
+		return ErrClosed
+	}
+	c.writeMu.Lock()
+	defer c.writeMu.Unlock()
+	if err := writeFrame(c.bw, true, opcode, payload); err != nil {
+		return err
+	}
+	return c.bw.Flush()
+}
+
+func toMsgType(opcode byte) MessageType {
+	if opcode == opText {
+		return TextMessage
+	}
+	return BinaryMessage
+}
+
+// closePayload builds a close frame payload (2-byte code + optional UTF-8 reason).
+func closePayload(code CloseCode, reason string) []byte {
+	if code == 0 {
+		return nil
+	}
+	// 2 bytes code + reason
+	buf := make([]byte, 2+len(reason))
+	binary.BigEndian.PutUint16(buf[:2], uint16(code))
+	copy(buf[2:], []byte(reason))
+	return buf
+}
+
+// parseClosePayload extracts code & reason from a close frame payload.
+func parseClosePayload(p []byte) (CloseCode, string) {
+	if len(p) < 2 {
+		return CloseNoStatusRcvd, ""
+	}
+	code := CloseCode(binary.BigEndian.Uint16(p[:2]))
+	reason := string(p[2:])
+	return code, reason
+}