Smux

https://github.com/xtaci/smux
Admin
Smux is a multiplexing library for Go that enables multiple logical streams over a single underlying...
Tokens:5,202
Snippets:43
Trust Score:9.4
License:MIT
Update:6 months ago
Show doc for...
Tokens:
Raw
### Smux Client Example

Source: https://github.com/xtaci/smux/blob/master/README.md

Demonstrates setting up a smux client to establish a connection, open a stream, send data, and close the session. Requires a net.Conn and the smux library.

```go
func client() {
    // Get a TCP connection
    conn, err := net.Dial(...) 
    if err != nil {
        panic(err)
    }

    // Setup client side of smux
    session, err := smux.Client(conn, nil)
    if err != nil {
        panic(err)
    }

    // Open a new stream
    stream, err := session.OpenStream()
    if err != nil {
        panic(err)
    }

    // Stream implements io.ReadWriteCloser
    stream.Write([]byte("ping"))
    stream.Close()
    session.Close()
}
```

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

### Smux Server Example

Source: https://github.com/xtaci/smux/blob/master/README.md

Illustrates setting up a smux server to accept connections, establish a session, accept streams, read data, and close the session. Requires an established listener and the smux library.

```go
func server() {
    // Accept a TCP connection
    conn, err := listener.Accept()
    if err != nil {
        panic(err)
    }

    // Setup server side of smux
    session, err := smux.Server(conn, nil)
    if err != nil {
        panic(err)
    }

    // Accept a stream
    stream, err := session.AcceptStream()
    if err != nil {
        panic(err)
    }

    // Listen for a message
    buf := make([]byte, 4)
    stream.Read(buf)
    stream.Close()
    session.Close()
}
```

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

### Implement Client-Server Communication

Source: https://context7.com/xtaci/smux/llms.txt

A complete example demonstrating bidirectional multiplexed communication using smux.Stream.

```go
package main

import (
    "fmt"
    "io"
    "net"
    "sync"
    "time"
    "github.com/xtaci/smux"
)

func server(wg *sync.WaitGroup) {
    defer wg.Done()

    listener, _ := net.Listen("tcp", ":8080")
    defer listener.Close()

    conn, _ := listener.Accept()

    config := smux.DefaultConfig()
    config.Version = 2 // Enable flow control
    session, _ := smux.Server(conn, config)
    defer session.Close()

    for i := 0; i < 3; i++ {
        stream, err := session.AcceptStream()
        if err != nil {
            return
        }

        go func(s *smux.Stream, id int) {
            defer s.Close()
            buf := make([]byte, 1024)
            n, _ := s.Read(buf)
            fmt.Printf("Server received on stream %d: %s\n", id, buf[:n])
            s.Write([]byte(fmt.Sprintf("Response from stream %d", id)))
        }(stream, i)
    }

    time.Sleep(time.Second)
}

func client() {
    time.Sleep(100 * time.Millisecond) // Wait for server

    conn, _ := net.Dial("tcp", "localhost:8080")

    config := smux.DefaultConfig()
    config.Version = 2
    session, _ := smux.Client(conn, config)
    defer session.Close()

    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()

            stream, err := session.OpenStream()
            if err != nil {
                return
            }
            defer stream.Close()

            stream.Write([]byte(fmt.Sprintf("Hello from client stream %d", id)))

            buf := make([]byte, 1024)
            n, _ := stream.Read(buf)
            fmt.Printf("Client received: %s\n", buf[:n])
        }(i)
    }
    wg.Wait()
}

func main() {
    var wg sync.WaitGroup
    wg.Add(1)
    go server(&wg)
    client()
    wg.Wait()
}

// Output:
// Server received on stream 0: Hello from client stream 0
// Server received on stream 1: Hello from client stream 1
// Server received on stream 2: Hello from client stream 2
// Client received: Response from stream 0
// Client received: Response from stream 1
// Client received: Response from stream 2
```

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

### Manage smux Session in Go

Source: https://context7.com/xtaci/smux/llms.txt

Provides examples of checking session status, retrieving connection information, and performing a graceful shutdown of an smux session. This code is for a client connecting to localhost:8080.

```go
package main

import (
    "fmt"
    "net"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Client(conn, nil)

    // Check if session is closed
    if session.IsClosed() {
        fmt.Println("Session is closed")
        return
    }

    // Get number of active streams
    numStreams := session.NumStreams()
    fmt.Printf("Active streams: %d\n", numStreams)

    // Get underlying connection addresses
    localAddr := session.LocalAddr()
    remoteAddr := session.RemoteAddr()
    fmt.Printf("Local: %v, Remote: %v\n", localAddr, remoteAddr)

    // Monitor session close with channel
    closeChan := session.CloseChan()
    go func() {
        <-closeChan
        fmt.Println("Session closed!")
    }()

    // Graceful shutdown - closes all streams and underlying connection
    err := session.Close()
    if err != nil {
        fmt.Printf("Close error: %v\n", err)
    }
}
```

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

### Create a server session

Source: https://github.com/xtaci/smux/blob/master/AGENTS.md

Initializes a new smux server session to accept incoming streams.

```go
conn, _ := listener.Accept()
session, _ := smux.Server(conn, nil)
stream, _ := session.AcceptStream()
```

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

### Create Smux Server Session and Handle Streams

Source: https://context7.com/xtaci/smux/llms.txt

Initialize a server-side multiplexed session to accept incoming connections. Handle new streams concurrently using goroutines and echo data back to the client.

```go
package main

import (
    "net"
    "github.com/xtaci/smux"
)

func main() {
    listener, err := net.Listen("tcp", ":8080")
    if err != nil {
        panic(err)
    }
    defer listener.Close()

    for {
        conn, err := listener.Accept()
        if err != nil {
            continue
        }

        // Create smux server session
        session, err := smux.Server(conn, nil)
        if err != nil {
            conn.Close()
            continue
        }

        go handleSession(session)
    }
}

func handleSession(session *smux.Session) {
    defer session.Close()

    for {
        stream, err := session.AcceptStream()
        if err != nil {
            return
        }
        go handleStream(stream)
    }
}

func handleStream(stream *smux.Stream) {
    defer stream.Close()
    buf := make([]byte, 1024)
    for {
        n, err := stream.Read(buf)
        if err != nil {
            return
        }
        stream.Write(buf[:n]) // Echo back
    }
}
```

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

### Run tests and benchmarks

Source: https://github.com/xtaci/smux/blob/master/AGENTS.md

Commands to execute the test suite and performance benchmarks for the project.

```bash
go test -v .
```

```bash
go test -v -run=^$ -bench .
```

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

### Accept Streams in Go

Source: https://context7.com/xtaci/smux/llms.txt

Shows how to accept incoming streams on the server side using smux. It includes setting an accept deadline and implementing a simple echo server. Requires a smux client to connect to localhost:8080.

```go
package main

import (
    "io"
    "net"
    "time"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Server(conn, nil)
    defer session.Close()

    // Set accept deadline (optional)
    session.SetDeadline(time.Now().Add(30 * time.Second))

    for {
        // Block until a new stream arrives
        stream, err := session.AcceptStream()
        if err == smux.ErrTimeout {
            continue // Timeout, try again
        }
        if err != nil {
            return // Session closed or error
        }

        go func(s *smux.Stream) {
            defer s.Close()
            io.Copy(s, s) // Echo server
        }(stream)
    }
}
```

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

### Create a client session

Source: https://github.com/xtaci/smux/blob/master/AGENTS.md

Initializes a new smux client session over an existing connection.

```go
conn, _ := net.Dial(...)
session, _ := smux.Client(conn, nil) // nil for default config
stream, _ := session.OpenStream()
```

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

### Create Smux Client Session

Source: https://context7.com/xtaci/smux/llms.txt

Initialize a new client-side multiplexed session over an existing network connection. Pass nil for default configuration or a custom Config for specific settings.

```go
package main

import (
    "net"
    "github.com/xtaci/smux"
)

func main() {
    // Establish TCP connection
    conn, err := net.Dial("tcp", "localhost:8080")
    if err != nil {
        panic(err)
    }
    defer conn.Close()

    // Create smux client session with default config (pass nil)
    session, err := smux.Client(conn, nil)
    if err != nil {
        panic(err)
    }
    defer session.Close()

    // Or with custom config for version 2 flow control
    config := smux.DefaultConfig()
    config.Version = 2
    sessionV2, err := smux.Client(conn, config)
    if err != nil {
        panic(err)
    }
    defer sessionV2.Close()
}
```

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

### Use Generic IO Interface

Source: https://context7.com/xtaci/smux/llms.txt

Utilize Open and Accept methods to return io.ReadWriteCloser for simplified integration with standard IO operations.

```go
package main

import (
    "io"
    "net"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Client(conn, nil)
    defer session.Close()

    // Open returns io.ReadWriteCloser instead of *Stream
    rwc, err := session.Open()
    if err != nil {
        panic(err)
    }
    defer rwc.Close()

    // Use standard io operations
    io.WriteString(rwc, "Hello via io.ReadWriteCloser")

    buf := make([]byte, 1024)
    n, _ := rwc.Read(buf)

    // On server side, Accept also returns io.ReadWriteCloser
    // rwc, err := session.Accept()
}
```

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

### Open a New Stream in Go

Source: https://context7.com/xtaci/smux/llms.txt

Demonstrates opening a new logical stream within an smux session and performing basic write and read operations. Ensure a smux server is listening on localhost:8080.

```go
package main

import (
    "fmt"
    "net"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Client(conn, nil)
    defer session.Close()

    // Open a new stream
    stream, err := session.OpenStream()
    if err != nil {
        panic(err)
    }
    defer stream.Close()

    // Stream implements net.Conn interface
    _, err = stream.Write([]byte("Hello, smux!"))
    if err != nil {
        panic(err)
    }

    // Read response
    buf := make([]byte, 1024)
    n, err := stream.Read(buf)
    if err != nil {
        panic(err)
    }
    fmt.Printf("Received: %s\n", buf[:n])

    // Open multiple concurrent streams
    for i := 0; i < 10; i++ {
        go func(id int) {
            s, err := session.OpenStream()
            if err != nil {
                return
            }
            defer s.Close()
            s.Write([]byte(fmt.Sprintf("Stream %d", id)))
        }(i)
    }
}
```

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

### Configure Smux Session Parameters

Source: https://context7.com/xtaci/smux/llms.txt

Use the Config struct to tune session parameters like protocol version, keepalive settings, frame sizes, and buffer limits. Verify the configuration using VerifyConfig before use.

```go
package main

import (
    "time"
    "github.com/xtaci/smux"
)

func main() {
    // Use default configuration
    defaultCfg := smux.DefaultConfig()
    // Output: Version=1, KeepAliveInterval=10s, KeepAliveTimeout=30s,
    //         MaxFrameSize=32768, MaxReceiveBuffer=4194304, MaxStreamBuffer=65536

    // Custom configuration with version 2 (flow control enabled)
    customCfg := &smux.Config{
        Version:           2,                    // Protocol version (1 or 2)
        KeepAliveDisabled: false,               // Enable keepalive
        KeepAliveInterval: 10 * time.Second,    // Send NOP every 10s
        KeepAliveTimeout:  30 * time.Second,    // Close if no data for 30s
        MaxFrameSize:      32768,               // Max 32KB per frame
        MaxReceiveBuffer:  4194304,             // 4MB session receive buffer
        MaxStreamBuffer:   65536,               // 64KB per-stream buffer
    }

    // Validate configuration before use
    if err := smux.VerifyConfig(customCfg); err != nil {
        panic(err)
    }
}
```

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

### Stream Read/Write with Deadlines in Go

Source: https://context7.com/xtaci/smux/llms.txt

Illustrates setting read and write deadlines on smux streams, performing half-close operations, and retrieving address information. This code is for a client connecting to localhost:8080.

```go
package main

import (
    "io"
    "net"
    "time"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Client(conn, nil)
    stream, _ := session.OpenStream()
    defer stream.Close()

    // Set read/write deadlines
    stream.SetDeadline(time.Now().Add(5 * time.Second))
    // Or set them individually
    stream.SetReadDeadline(time.Now().Add(5 * time.Second))
    stream.SetWriteDeadline(time.Now().Add(5 * time.Second))

    // Write data
    data := []byte("Request data")
    n, err := stream.Write(data)
    if err == smux.ErrTimeout {
        // Handle timeout
    }

    // Read data
    buf := make([]byte, 4096)
    n, err = stream.Read(buf)
    if err == io.EOF {
        // Remote closed the stream
    }

    // Half-close: close write side but continue reading
    err = stream.CloseWrite()
    if err != nil {
        panic(err)
    }
    // Can still read from stream after CloseWrite
    n, err = stream.Read(buf)

    // Get local and remote addresses
    localAddr := stream.LocalAddr()   // net.Addr or nil
    remoteAddr := stream.RemoteAddr() // net.Addr or nil

    // Use WriteTo for efficient streaming
    var writer io.Writer // Your destination writer
    written, err := stream.WriteTo(writer)
}
```

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

### Smux Go Benchmark Output

Source: https://github.com/xtaci/smux/blob/master/README.md

Output from running Go benchmarks for the Smux library, showing performance metrics for various operations like msb, AcceptClose, ConnSmux, and ConnTCP.

```bash
$ go test -v -run=^$ -bench .
goos: darwin
goarch: amd64
pkg: github.com/xtaci/smux
BenchmarkMSB-4           	30000000	        51.8 ns/op
BenchmarkAcceptClose-4   	   50000	     36783 ns/op
BenchmarkConnSmux-4      	   30000	     58335 ns/op	2246.88 MB/s	    1208 B/op	      19 allocs/op
BenchmarkConnTCP-4       	   50000	     25579 ns/op	5124.04 MB/s	       0 B/op	       0 allocs/op
PASS
ok  	github.com/xtaci/smux	7.811s
```

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

### Handle Smux Errors

Source: https://context7.com/xtaci/smux/llms.txt

Identify specific failure scenarios using Smux error types and standard Go error checking.

```go
package main

import (
    "errors"
    "io"
    "net"
    "github.com/xtaci/smux"
)

func main() {
    conn, _ := net.Dial("tcp", "localhost:8080")
    session, _ := smux.Client(conn, nil)

    stream, err := session.OpenStream()
    if err != nil {
        switch {
        case err == io.ErrClosedPipe:
            // Session is closed
        case err == smux.ErrGoAway:
            // Stream ID overflow, need new connection
        case errors.Is(err, smux.ErrTimeout):
            // Operation timed out
        case err == smux.ErrInvalidProtocol:
            // Protocol version mismatch
        case err == smux.ErrConsumed:
            // Flow control error (v2 only)
        case err == smux.ErrWouldBlock:
            // Non-blocking operation would block
        }
        return
    }
    defer stream.Close()

    // Read/Write errors
    buf := make([]byte, 1024)
    _, err = stream.Read(buf)
    if err != nil {
        if err == io.EOF {
            // Stream closed normally by remote
        }
        if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
            // Deadline exceeded
        }
    }
}
```

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

### Smux Frame Wire Format

Source: https://github.com/xtaci/smux/blob/master/README.md

Defines the byte-level structure of the Smux frame, including fields for version, command, length, stream ID, and data, along with the valid values for command types.

```text
 +---------------+---------------+-------------------------------+
 |  VERSION (1B) |    CMD (1B)   |          LENGTH (2B)          |
 +---------------+---------------+-------------------------------+
 |                          STREAMID (4B)                        |
 +---------------------------------------------------------------+
 |                                                               |
 /                        DATA (Variable)                        /
 |                                                               |
 +---------------------------------------------------------------+

VALUES FOR LATEST VERSION:
VERSION:
    1/2
    
CMD:
    cmdSYN(0)
    cmdFIN(1)
    cmdPSH(2)
    cmdNOP(3)
    cmdUPD(4)	// only supported on version 2
    
STREAMID:
    client use odd numbers starts from 1
    server use even numbers starts from 0
    
cmdUPD:
    | CONSUMED(4B) | WINDOW(4B) |
```
Tokens:
Raw