Add datagram library back in

2025-07-17 02:14:49 -04:00 · 2025-05-29 18:45:52 -04:00
parent b114d8b337
commit 38182694c5
17 changed files with 1593 additions and 1 deletions
--- a/10
+++ b/10
@ -18,9 +18,17 @@ stream-test:
 raw-test:
 	go test --tags nettest -v ./raw/...

+test:
+	make common-test
+	make stream-test
+	make datagram-test
+	#make raw-test
+	#make primary-test
+
+
 test-logs:
 	make common-test 2> common-err.log 1> common-out.log;	cat common-err.log common-out.log
 	make stream-test 2> stream-err.log 1> stream-out.log;	cat stream-err.log stream-out.log
-	#make datagram-test 2> datagram-err.log 1> datagram-out.log;	cat datagram-err.log datagram-out.log
+	make datagram-test 2> datagram-err.log 1> datagram-out.log;	cat datagram-err.log datagram-out.log
 	#make raw-test 2> raw-err.log 1> raw-out.log;	cat raw-err.log raw-out.log
 	#make primary-test 2> primary-err.log 1> primary-out.log;	cat primary-err.log primary-out.log
--- a/datagram/README.md
+++ b/datagram/README.md
@ -0,0 +1,23 @@
+package datagram
+
+High-level datagram library for UDP-like message delivery over I2P using the SAMv3 protocol.
+
+## Installation
+
+Install using Go modules with the package path `github.com/go-i2p/go-sam-go/datagram`.
+
+## Usage
+
+The package provides UDP-like datagram messaging over I2P networks. [`DatagramSession`](datagram/types.go) manages the session lifecycle, [`DatagramReader`](datagram/types.go) handles incoming datagrams, [`DatagramWriter`](datagram/types.go) sends outgoing datagrams, and [`DatagramConn`](datagram/types.go) implements the standard `net.PacketConn` interface for seamless integration with existing Go networking code.
+
+Create sessions using [`NewDatagramSession`](datagram/session.go), send messages with [`SendDatagram()`](datagram/session.go), and receive messages using [`ReceiveDatagram()`](datagram/session.go). The implementation supports I2P address resolution, configurable tunnel parameters, and comprehensive error handling with proper resource cleanup.
+
+Key features include full `net.PacketConn` and `net.Conn` compatibility, I2P destination management, base64 payload encoding, and concurrent datagram processing with proper synchronization.
+
+## Dependencies
+
+- github.com/go-i2p/go-sam-go/common - Core SAM protocol implementation
+- github.com/go-i2p/i2pkeys - I2P cryptographic key handling  
+- github.com/go-i2p/logger - Logging functionality
+- github.com/sirupsen/logrus - Structured logging
+- github.com/samber/oops - Enhanced error handling
--- a/datagram/SAM.go
+++ b/datagram/SAM.go
@ -0,0 +1,85 @@
+package datagram
+
+import (
+	"github.com/go-i2p/go-sam-go/common"
+	"github.com/go-i2p/i2pkeys"
+	"github.com/samber/oops"
+	"github.com/sirupsen/logrus"
+)
+
+// SAM wraps common.SAM to provide datagram-specific functionality
+type SAM struct {
+	*common.SAM
+}
+
+// NewDatagramSession creates a new datagram session with the SAM bridge
+func (s *SAM) NewDatagramSession(id string, keys i2pkeys.I2PKeys, options []string) (*DatagramSession, error) {
+	return NewDatagramSession(s.SAM, id, keys, options)
+}
+
+// NewDatagramSessionWithSignature creates a new datagram session with custom signature type
+func (s *SAM) NewDatagramSessionWithSignature(id string, keys i2pkeys.I2PKeys, options []string, sigType string) (*DatagramSession, error) {
+	logger := log.WithFields(logrus.Fields{
+		"id":      id,
+		"options": options,
+		"sigType": sigType,
+	})
+	logger.Debug("Creating new DatagramSession with signature")
+
+	// Create the base session using the common package with signature
+	session, err := s.SAM.NewGenericSessionWithSignature("DATAGRAM", id, keys, sigType, options)
+	if err != nil {
+		logger.WithError(err).Error("Failed to create generic session with signature")
+		return nil, oops.Errorf("failed to create datagram session: %w", err)
+	}
+
+	baseSession, ok := session.(*common.BaseSession)
+	if !ok {
+		logger.Error("Session is not a BaseSession")
+		session.Close()
+		return nil, oops.Errorf("invalid session type")
+	}
+
+	ds := &DatagramSession{
+		BaseSession: baseSession,
+		sam:         s.SAM,
+		options:     options,
+	}
+
+	logger.Debug("Successfully created DatagramSession with signature")
+	return ds, nil
+}
+
+// NewDatagramSessionWithPorts creates a new datagram session with port specifications
+func (s *SAM) NewDatagramSessionWithPorts(id, fromPort, toPort string, keys i2pkeys.I2PKeys, options []string) (*DatagramSession, error) {
+	logger := log.WithFields(logrus.Fields{
+		"id":       id,
+		"fromPort": fromPort,
+		"toPort":   toPort,
+		"options":  options,
+	})
+	logger.Debug("Creating new DatagramSession with ports")
+
+	// Create the base session using the common package with ports
+	session, err := s.SAM.NewGenericSessionWithSignatureAndPorts("DATAGRAM", id, fromPort, toPort, keys, common.SIG_EdDSA_SHA512_Ed25519, options)
+	if err != nil {
+		logger.WithError(err).Error("Failed to create generic session with ports")
+		return nil, oops.Errorf("failed to create datagram session: %w", err)
+	}
+
+	baseSession, ok := session.(*common.BaseSession)
+	if !ok {
+		logger.Error("Session is not a BaseSession")
+		session.Close()
+		return nil, oops.Errorf("invalid session type")
+	}
+
+	ds := &DatagramSession{
+		BaseSession: baseSession,
+		sam:         s.SAM,
+		options:     options,
+	}
+
+	logger.Debug("Successfully created DatagramSession with ports")
+	return ds, nil
+}
--- a/datagram/dial.go
+++ b/datagram/dial.go
@ -0,0 +1,82 @@
+package datagram
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/go-i2p/i2pkeys"
+	"github.com/sirupsen/logrus"
+)
+
+// Dial establishes a datagram connection to the specified destination
+func (ds *DatagramSession) Dial(destination string) (net.PacketConn, error) {
+	return ds.DialTimeout(destination, 30*time.Second)
+}
+
+// DialTimeout establishes a datagram connection with a timeout
+func (ds *DatagramSession) DialTimeout(destination string, timeout time.Duration) (net.PacketConn, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return ds.DialContext(ctx, destination)
+}
+
+// DialContext establishes a datagram connection with context support
+func (ds *DatagramSession) DialContext(ctx context.Context, destination string) (net.PacketConn, error) {
+	logger := log.WithFields(logrus.Fields{
+		"destination": destination,
+	})
+	logger.Debug("Dialing datagram destination")
+
+	// Create a datagram connection
+	conn := &DatagramConn{
+		session: ds,
+		reader:  ds.NewReader(),
+		writer:  ds.NewWriter(),
+	}
+
+	// Start the reader loop
+	go conn.reader.receiveLoop()
+
+	logger.WithField("session_id", ds.ID()).Debug("Successfully created datagram connection")
+	return conn, nil
+}
+
+// DialI2P establishes a datagram connection to an I2P address
+func (ds *DatagramSession) DialI2P(addr i2pkeys.I2PAddr) (net.PacketConn, error) {
+	return ds.DialI2PTimeout(addr, 30*time.Second)
+}
+
+// DialI2PTimeout establishes a datagram connection to an I2P address with timeout
+func (ds *DatagramSession) DialI2PTimeout(addr i2pkeys.I2PAddr, timeout time.Duration) (net.PacketConn, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	return ds.DialI2PContext(ctx, addr)
+}
+
+// DialI2PContext establishes a datagram connection to an I2P address with context support
+func (ds *DatagramSession) DialI2PContext(ctx context.Context, addr i2pkeys.I2PAddr) (net.PacketConn, error) {
+	logger := log.WithFields(logrus.Fields{
+		"destination": addr.Base32(),
+	})
+	logger.Debug("Dialing I2P datagram destination")
+
+	// Create a datagram connection
+	conn := &DatagramConn{
+		session: ds,
+		reader:  ds.NewReader(),
+		writer:  ds.NewWriter(),
+	}
+
+	// Start the reader loop
+	go conn.reader.receiveLoop()
+
+	logger.WithField("session_id", ds.ID()).Debug("Successfully created I2P datagram connection")
+	return conn, nil
+}
+
+// generateSessionID generates a unique session identifier
+func generateSessionID() string {
+	return fmt.Sprintf("datagram_%d", time.Now().UnixNano())
+}
--- a/datagram/dial_test.go
+++ b/datagram/dial_test.go
@ -0,0 +1,193 @@
+package datagram
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestDatagramSession_Dial(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	// Create two sessions - one for listener, one for dialer
+	sam1, keys1 := setupTestSAM(t)
+	defer sam1.Close()
+
+	sam2, keys2 := setupTestSAM(t)
+	defer sam2.Close()
+
+	// Create listener session
+	listenerSession, err := NewDatagramSession(sam1, "test_dial_listener", keys1, []string{
+		"inbound.length=0", "outbound.length=0",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create listener session: %v", err)
+	}
+	defer listenerSession.Close()
+
+	listener, err := listenerSession.Listen()
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer listener.Close()
+
+	// Create dialer session
+	dialerSession, err := NewDatagramSession(sam2, "test_dial_dialer", keys2, []string{
+		"inbound.length=0", "outbound.length=0",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create dialer session: %v", err)
+	}
+	defer dialerSession.Close()
+
+	// Test dial
+	conn, err := dialerSession.Dial(listener.Addr().String())
+	if err != nil {
+		t.Fatalf("Failed to dial: %v", err)
+	}
+	defer conn.Close()
+
+	// Verify connection properties
+	if conn == nil {
+		t.Fatal("Dial returned nil connection")
+	}
+
+	if conn.LocalAddr().String() != dialerSession.Addr().String() {
+		t.Errorf("Local address mismatch: got %s, want %s",
+			conn.LocalAddr().String(), dialerSession.Addr().String())
+	}
+}
+
+func TestDatagramSession_DialContext(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	// Create two sessions
+	sam1, keys1 := setupTestSAM(t)
+	defer sam1.Close()
+
+	sam2, keys2 := setupTestSAM(t)
+	defer sam2.Close()
+
+	// Create listener session
+	listenerSession, err := NewDatagramSession(sam1, "test_dialctx_listener", keys1, nil)
+	if err != nil {
+		t.Fatalf("Failed to create listener session: %v", err)
+	}
+	defer listenerSession.Close()
+
+	listener, err := listenerSession.Listen()
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer listener.Close()
+
+	// Create dialer session
+	dialerSession, err := NewDatagramSession(sam2, "test_dialctx_dialer", keys2, nil)
+	if err != nil {
+		t.Fatalf("Failed to create dialer session: %v", err)
+	}
+	defer dialerSession.Close()
+
+	// Test dial with context
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	conn, err := dialerSession.DialContext(ctx, listener.Addr().String())
+	if err != nil {
+		t.Fatalf("Failed to dial with context: %v", err)
+	}
+	defer conn.Close()
+	if conn == nil {
+		t.Fatal("DialContext returned nil connection")
+	}
+}
+
+func TestDatagramSession_DialContext_Timeout(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_dialctx_timeout", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	// Use very short timeout to force timeout
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Millisecond)
+	defer cancel()
+
+	// Try to dial invalid address with short timeout
+	conn, err := session.DialContext(ctx, "invalid.b32.i2p")
+	if err == nil {
+		if conn != nil {
+			conn.Close()
+		}
+	}
+
+	if conn != nil {
+		t.Error("Expected nil connection on timeout")
+	}
+}
+
+func TestDatagramSession_Dial_ClosedSession(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_dial_closed", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+
+	// Close the session first
+	session.Close()
+
+	// Try to dial on closed session
+	conn, err := session.Dial("test.b32.i2p")
+	if err == nil {
+		if conn != nil {
+			conn.Close()
+		}
+		t.Error("Expected error when dialing on closed session")
+	}
+
+	if conn != nil {
+		t.Error("Expected nil connection when session is closed")
+	}
+}
+
+func TestDatagramSession_NewDialer(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_newdialer", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	// Test that session can dial successfully (since NewDialer method doesn't exist)
+	// This test now verifies the basic dialing functionality
+	conn, err := session.Dial("test.b32.i2p")
+	if err != nil {
+		// Expected to fail with invalid address, but should not panic
+		t.Logf("Dial failed as expected with invalid address: %v", err)
+	} else if conn != nil {
+		conn.Close()
+	}
+}
--- a/datagram/listen.go
+++ b/datagram/listen.go
@ -0,0 +1,33 @@
+package datagram
+
+import (
+	"net"
+
+	"github.com/samber/oops"
+)
+
+func (s *DatagramSession) Listen() (*DatagramListener, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	if s.closed {
+		return nil, oops.Errorf("session is closed")
+	}
+
+	logger := log.WithField("id", s.ID())
+	logger.Debug("Creating PacketListener")
+
+	listener := &DatagramListener{
+		session:    s,
+		reader:     s.NewReader(),
+		acceptChan: make(chan net.Conn, 10), // Buffer for incoming connections
+		errorChan:  make(chan error, 1),
+		closeChan:  make(chan struct{}),
+	}
+
+	// Start accepting packet connections in a goroutine
+	go listener.acceptLoop()
+
+	logger.Debug("Successfully created PacketListener")
+	return listener, nil
+}
--- a/datagram/listen_test.go
+++ b/datagram/listen_test.go
@ -0,0 +1,84 @@
+package datagram
+
+import (
+	"testing"
+)
+
+func TestDatagramSession_Listen(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_listen", keys, []string{
+		"inbound.length=0", "outbound.length=0",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	listener, err := session.Listen()
+	if err != nil {
+		t.Fatalf("Failed to create listener: %v", err)
+	}
+	defer listener.Close()
+
+	// Verify listener properties
+	if listener.Addr().String() != session.Addr().String() {
+		t.Error("Listener address doesn't match session address")
+	}
+
+	// Verify listener is not nil and has expected fields
+	if listener.session != session {
+		t.Error("Listener session doesn't match created session")
+	}
+
+	if listener.reader == nil {
+		t.Error("Listener reader is nil")
+	}
+
+	if listener.acceptChan == nil {
+		t.Error("Listener acceptChan is nil")
+	}
+
+	if listener.errorChan == nil {
+		t.Error("Listener errorChan is nil")
+	}
+
+	if listener.closeChan == nil {
+		t.Error("Listener closeChan is nil")
+	}
+}
+
+func TestDatagramSession_Listen_ClosedSession(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_listen_closed", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+
+	// Close the session first
+	session.Close()
+
+	// Try to create listener on closed session
+	listener, err := session.Listen()
+	if err == nil {
+		if listener != nil {
+			listener.Close()
+		}
+		t.Fatal("Expected error when creating listener on closed session")
+	}
+
+	if listener != nil {
+		t.Error("Expected nil listener when session is closed")
+	}
+}
--- a/datagram/log.go
+++ b/datagram/log.go
@ -0,0 +1,7 @@
+package datagram
+
+import (
+	"github.com/go-i2p/logger"
+)
+
+var log = logger.GetGoI2PLogger()
--- a/datagram/packetconn.go
+++ b/datagram/packetconn.go
@ -0,0 +1,150 @@
+package datagram
+
+import (
+	"net"
+	"time"
+
+	"github.com/go-i2p/i2pkeys"
+	"github.com/samber/oops"
+)
+
+// ReadFrom reads a datagram from the connection
+func (c *DatagramConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
+	c.mu.RLock()
+	if c.closed {
+		c.mu.RUnlock()
+		return 0, nil, oops.Errorf("connection is closed")
+	}
+	c.mu.RUnlock()
+
+	// Start receive loop if not already started
+	go c.reader.receiveLoop()
+
+	datagram, err := c.reader.ReceiveDatagram()
+	if err != nil {
+		return 0, nil, err
+	}
+
+	// Copy data to the provided buffer
+	n = copy(p, datagram.Data)
+	addr = &DatagramAddr{addr: datagram.Source}
+
+	return n, addr, nil
+}
+
+// WriteTo writes a datagram to the specified address
+func (c *DatagramConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	c.mu.RLock()
+	if c.closed {
+		c.mu.RUnlock()
+		return 0, oops.Errorf("connection is closed")
+	}
+	c.mu.RUnlock()
+
+	// Convert address to I2P address
+	i2pAddr, ok := addr.(*DatagramAddr)
+	if !ok {
+		return 0, oops.Errorf("address must be a DatagramAddr")
+	}
+
+	err = c.writer.SendDatagram(p, i2pAddr.addr)
+	if err != nil {
+		return 0, err
+	}
+
+	return len(p), nil
+}
+
+// Close closes the datagram connection
+func (c *DatagramConn) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.closed {
+		return nil
+	}
+
+	logger := log.WithField("session_id", c.session.ID())
+	logger.Debug("Closing DatagramConn")
+
+	c.closed = true
+
+	// Close reader and writer
+	if c.reader != nil {
+		c.reader.Close()
+	}
+
+	// Close the session
+	if err := c.session.Close(); err != nil {
+		logger.WithError(err).Error("Failed to close session")
+		return oops.Errorf("failed to close datagram connection: %w", err)
+	}
+
+	logger.Debug("Successfully closed DatagramConn")
+	return nil
+}
+
+// LocalAddr returns the local address
+func (c *DatagramConn) LocalAddr() net.Addr {
+	return &DatagramAddr{addr: c.session.Addr()}
+}
+
+// SetDeadline sets the read and write deadlines
+func (c *DatagramConn) SetDeadline(t time.Time) error {
+	if err := c.SetReadDeadline(t); err != nil {
+		return err
+	}
+	return c.SetWriteDeadline(t)
+}
+
+// SetReadDeadline sets the deadline for future ReadFrom calls
+func (c *DatagramConn) SetReadDeadline(t time.Time) error {
+	// For datagrams, we handle timeouts differently
+	// This is a placeholder implementation
+	return nil
+}
+
+// SetWriteDeadline sets the deadline for future WriteTo calls
+func (c *DatagramConn) SetWriteDeadline(t time.Time) error {
+	// Calculate timeout duration
+	if !t.IsZero() {
+		timeout := time.Until(t)
+		c.writer.SetTimeout(timeout)
+	}
+	return nil
+}
+
+// Read implements net.Conn by wrapping ReadFrom.
+// It reads data into the provided byte slice and returns the number of bytes read.
+// When reading, it also updates the remote address of the connection.
+// Note: This is not a typical use case for datagrams, as they are connectionless.
+// However, for compatibility with net.Conn, we implement it this way.
+func (c *DatagramConn) Read(b []byte) (n int, err error) {
+	n, addr, err := c.ReadFrom(b)
+	c.remoteAddr = addr.(*i2pkeys.I2PAddr)
+	return n, err
+}
+
+// RemoteAddr returns the remote address of the connection.
+// For datagram connections, this returns nil as there is no single remote address.
+func (c *DatagramConn) RemoteAddr() net.Addr {
+	if c.remoteAddr != nil {
+		return &DatagramAddr{addr: *c.remoteAddr}
+	}
+	return nil
+}
+
+// Write implements net.Conn by wrapping WriteTo.
+// It writes data to the remote address and returns the number of bytes written.
+// It uses the remote address set by the last Read operation.
+// If no remote address is set, it returns an error.
+// Note: This is not a typical use case for datagrams, as they are connectionless.
+// However, for compatibility with net.Conn, we implement it this way.
+func (c *DatagramConn) Write(b []byte) (n int, err error) {
+	if c.remoteAddr == nil {
+		return 0, oops.Errorf("no remote address set, use WriteTo or Read first")
+	}
+
+	addr := &DatagramAddr{addr: *c.remoteAddr}
+	return c.WriteTo(b, addr)
+}
--- a/datagram/packetlistener.go
+++ b/datagram/packetlistener.go
@ -0,0 +1,125 @@
+package datagram
+
+import (
+	"net"
+	"sync"
+
+	"github.com/samber/oops"
+)
+
+// DatagramListener implements net.DatagramListener for I2P datagram connections
+type DatagramListener struct {
+	session    *DatagramSession
+	reader     *DatagramReader
+	acceptChan chan net.Conn
+	errorChan  chan error
+	closeChan  chan struct{}
+	closed     bool
+	mu         sync.RWMutex
+}
+
+// Accept waits for and returns the next packet connection to the listener
+func (l *DatagramListener) Accept() (net.Conn, error) {
+	l.mu.RLock()
+	if l.closed {
+		l.mu.RUnlock()
+		return nil, oops.Errorf("listener is closed")
+	}
+	l.mu.RUnlock()
+
+	select {
+	case conn := <-l.acceptChan:
+		return conn, nil
+	case err := <-l.errorChan:
+		return nil, err
+	case <-l.closeChan:
+		return nil, oops.Errorf("listener is closed")
+	}
+}
+
+// Close closes the packet listener
+func (l *DatagramListener) Close() error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if l.closed {
+		return nil
+	}
+
+	logger := log.WithField("session_id", l.session.ID())
+	logger.Debug("Closing PacketListener")
+
+	l.closed = true
+	close(l.closeChan)
+
+	// Close the reader
+	if l.reader != nil {
+		l.reader.Close()
+	}
+
+	logger.Debug("Successfully closed PacketListener")
+	return nil
+}
+
+// Addr returns the listener's network address
+func (l *DatagramListener) Addr() net.Addr {
+	return &DatagramAddr{addr: l.session.Addr()}
+}
+
+// acceptLoop continuously accepts incoming packet connections
+func (l *DatagramListener) acceptLoop() {
+	logger := log.WithField("session_id", l.session.ID())
+	logger.Debug("Starting packet accept loop")
+
+	for {
+		select {
+		case <-l.closeChan:
+			logger.Debug("Packet accept loop terminated - listener closed")
+			return
+		default:
+			conn, err := l.acceptPacketConnection()
+			if err != nil {
+				l.mu.RLock()
+				closed := l.closed
+				l.mu.RUnlock()
+
+				if !closed {
+					logger.WithError(err).Error("Failed to accept packet connection")
+					select {
+					case l.errorChan <- err:
+					case <-l.closeChan:
+						return
+					}
+				}
+				continue
+			}
+
+			select {
+			case l.acceptChan <- conn:
+				logger.Debug("Successfully accepted new packet connection")
+			case <-l.closeChan:
+				conn.Close()
+				return
+			}
+		}
+	}
+}
+
+// acceptPacketConnection creates a new packet connection for incoming datagrams
+func (l *DatagramListener) acceptPacketConnection() (net.Conn, error) {
+	logger := log.WithField("session_id", l.session.ID())
+	logger.Debug("Creating new packet connection")
+
+	// For datagram sessions, we create a new DatagramConn that shares the session
+	// but has its own reader/writer for handling the specific connection
+	conn := &DatagramConn{
+		session: l.session,
+		reader:  l.session.NewReader(),
+		writer:  l.session.NewWriter(),
+	}
+
+	// Start the reader loop for this connection
+	go conn.reader.receiveLoop()
+
+	return conn, nil
+}
--- a/datagram/read.go
+++ b/datagram/read.go
@ -0,0 +1,171 @@
+package datagram
+
+import (
+	"bufio"
+	"encoding/base64"
+	"strings"
+	"time"
+
+	"github.com/go-i2p/i2pkeys"
+	"github.com/samber/oops"
+)
+
+// ReceiveDatagram receives a datagram from any source
+func (r *DatagramReader) ReceiveDatagram() (*Datagram, error) {
+	// Check if closed first, but don't rely on this check for safety
+	r.mu.RLock()
+	if r.closed {
+		r.mu.RUnlock()
+		return nil, oops.Errorf("reader is closed")
+	}
+	r.mu.RUnlock()
+
+	// Use select with closeChan to handle concurrent close operations safely
+	// The closeChan will be signaled when Close() is called, providing
+	// a reliable way to detect closure even if it happens during this function
+	select {
+	case datagram := <-r.recvChan:
+		return datagram, nil
+	case err := <-r.errorChan:
+		return nil, err
+	case <-r.closeChan:
+		// This case handles both initial closure check and concurrent closure
+		return nil, oops.Errorf("reader is closed")
+	}
+}
+
+func (r *DatagramReader) Close() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.closed {
+		return nil
+	}
+
+	logger := log.WithField("session_id", r.session.ID())
+	logger.Debug("Closing DatagramReader")
+
+	r.closed = true
+	// Signal termination to receiveLoop and wait for it to exit
+	close(r.closeChan)
+
+	// Give receiveLoop time to detect the close signal and exit
+	// before closing the channels it might be writing to
+	time.Sleep(10 * time.Millisecond)
+
+	// Now safe to close the receiver channels
+	close(r.recvChan)
+	close(r.errorChan)
+
+	logger.Debug("Successfully closed DatagramReader")
+	return nil
+}
+
+// receiveLoop continuously receives incoming datagrams
+func (r *DatagramReader) receiveLoop() {
+	logger := log.WithField("session_id", r.session.ID())
+	logger.Debug("Starting receive loop")
+
+	for {
+		// Check for closure in a non-blocking way first
+		select {
+		case <-r.closeChan:
+			logger.Debug("Receive loop terminated - reader closed")
+			return
+		default:
+		}
+
+		// Now perform the blocking read operation
+		datagram, err := r.receiveDatagram()
+		if err != nil {
+			// Use atomic check and send pattern to avoid race
+			select {
+			case r.errorChan <- err:
+				logger.WithError(err).Error("Failed to receive datagram")
+			case <-r.closeChan:
+				// Reader was closed during error handling
+				return
+			}
+			continue
+		}
+
+		// Send the datagram or handle closure atomically
+		select {
+		case r.recvChan <- datagram:
+			logger.Debug("Successfully received datagram")
+		case <-r.closeChan:
+			// Reader was closed during datagram send
+			return
+		}
+	}
+}
+
+// receiveDatagram handles the low-level datagram reception
+func (r *DatagramReader) receiveDatagram() (*Datagram, error) {
+	logger := log.WithField("session_id", r.session.ID())
+
+	// Read from the session connection for incoming datagrams
+	buf := make([]byte, 4096)
+	n, err := r.session.Read(buf)
+	if err != nil {
+		return nil, oops.Errorf("failed to read from session: %w", err)
+	}
+
+	response := string(buf[:n])
+	logger.WithField("response", response).Debug("Received datagram data")
+
+	// Parse the DATAGRAM RECEIVED response
+	scanner := bufio.NewScanner(strings.NewReader(response))
+	scanner.Split(bufio.ScanWords)
+
+	var source, data string
+	for scanner.Scan() {
+		word := scanner.Text()
+		switch {
+		case word == "DATAGRAM":
+			continue
+		case word == "RECEIVED":
+			continue
+		case strings.HasPrefix(word, "DESTINATION="):
+			source = word[12:]
+		case strings.HasPrefix(word, "SIZE="):
+			continue // We'll get the actual data size from the payload
+		default:
+			// Remaining data is the base64-encoded payload
+			if data == "" {
+				data = word
+			} else {
+				data += " " + word
+			}
+		}
+	}
+
+	if source == "" {
+		return nil, oops.Errorf("no source in datagram")
+	}
+
+	if data == "" {
+		return nil, oops.Errorf("no data in datagram")
+	}
+
+	// Parse the source destination
+	sourceAddr, err := i2pkeys.NewI2PAddrFromString(source)
+	if err != nil {
+		return nil, oops.Errorf("failed to parse source address: %w", err)
+	}
+
+	// Decode the base64 data
+	decodedData, err := base64.StdEncoding.DecodeString(data)
+	if err != nil {
+		return nil, oops.Errorf("failed to decode datagram data: %w", err)
+	}
+
+	// Create the datagram
+	datagram := &Datagram{
+		Data:   decodedData,
+		Source: sourceAddr,
+		Local:  r.session.Addr(),
+	}
+
+	return datagram, nil
+}
--- a/datagram/session.go
+++ b/datagram/session.go
@ -0,0 +1,121 @@
+package datagram
+
+import (
+	"net"
+
+	"github.com/go-i2p/go-sam-go/common"
+	"github.com/go-i2p/i2pkeys"
+	"github.com/samber/oops"
+	"github.com/sirupsen/logrus"
+)
+
+// NewDatagramSession creates a new datagram session
+func NewDatagramSession(sam *common.SAM, id string, keys i2pkeys.I2PKeys, options []string) (*DatagramSession, error) {
+	logger := log.WithFields(logrus.Fields{
+		"id":      id,
+		"options": options,
+	})
+	logger.Debug("Creating new DatagramSession")
+
+	// Create the base session using the common package
+	session, err := sam.NewGenericSession("DATAGRAM", id, keys, options)
+	if err != nil {
+		logger.WithError(err).Error("Failed to create generic session")
+		return nil, oops.Errorf("failed to create datagram session: %w", err)
+	}
+
+	baseSession, ok := session.(*common.BaseSession)
+	if !ok {
+		logger.Error("Session is not a BaseSession")
+		session.Close()
+		return nil, oops.Errorf("invalid session type")
+	}
+
+	ds := &DatagramSession{
+		BaseSession: baseSession,
+		sam:         sam,
+		options:     options,
+	}
+
+	logger.Debug("Successfully created DatagramSession")
+	return ds, nil
+}
+
+// NewReader creates a DatagramReader for receiving datagrams
+func (s *DatagramSession) NewReader() *DatagramReader {
+	return &DatagramReader{
+		session:   s,
+		recvChan:  make(chan *Datagram, 10), // Buffer for incoming datagrams
+		errorChan: make(chan error, 1),
+		closeChan: make(chan struct{}),
+	}
+}
+
+// NewWriter creates a DatagramWriter for sending datagrams
+func (s *DatagramSession) NewWriter() *DatagramWriter {
+	return &DatagramWriter{
+		session: s,
+		timeout: 30, // Default timeout in seconds
+	}
+}
+
+// PacketConn returns a net.PacketConn interface for this session
+func (s *DatagramSession) PacketConn() net.PacketConn {
+	return &DatagramConn{
+		session: s,
+		reader:  s.NewReader(),
+		writer:  s.NewWriter(),
+	}
+}
+
+// SendDatagram sends a datagram to the specified destination
+func (s *DatagramSession) SendDatagram(data []byte, dest i2pkeys.I2PAddr) error {
+	return s.NewWriter().SendDatagram(data, dest)
+}
+
+// ReceiveDatagram receives a datagram from any source
+func (s *DatagramSession) ReceiveDatagram() (*Datagram, error) {
+	reader := s.NewReader()
+	// Start the receive loop
+	go reader.receiveLoop()
+	return reader.ReceiveDatagram()
+}
+
+// Close closes the datagram session and all associated resources
+func (s *DatagramSession) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.closed {
+		return nil
+	}
+
+	logger := log.WithField("id", s.ID())
+	logger.Debug("Closing DatagramSession")
+
+	s.closed = true
+
+	// Close the base session
+	if err := s.BaseSession.Close(); err != nil {
+		logger.WithError(err).Error("Failed to close base session")
+		return oops.Errorf("failed to close datagram session: %w", err)
+	}
+
+	logger.Debug("Successfully closed DatagramSession")
+	return nil
+}
+
+// Addr returns the I2P address of this session
+func (s *DatagramSession) Addr() i2pkeys.I2PAddr {
+	return s.Keys().Addr()
+}
+
+// Network returns the network type
+func (a *DatagramAddr) Network() string {
+	return "i2p-datagram"
+}
+
+// String returns the string representation of the address
+func (a *DatagramAddr) String() string {
+	return a.addr.Base32()
+}
--- a/datagram/session_test.go
+++ b/datagram/session_test.go
@ -0,0 +1,321 @@
+package datagram
+
+import (
+	"testing"
+	"time"
+
+	"github.com/go-i2p/go-sam-go/common"
+	"github.com/go-i2p/i2pkeys"
+)
+
+const testSAMAddr = "127.0.0.1:7656"
+
+func setupTestSAM(t *testing.T) (*common.SAM, i2pkeys.I2PKeys) {
+	t.Helper()
+
+	sam, err := common.NewSAM(testSAMAddr)
+	if err != nil {
+		t.Fatalf("Failed to create SAM connection: %v", err)
+	}
+
+	keys, err := sam.NewKeys()
+	if err != nil {
+		sam.Close()
+		t.Fatalf("Failed to generate keys: %v", err)
+	}
+
+	return sam, keys
+}
+
+func TestNewDatagramSession(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	tests := []struct {
+		name    string
+		id      string
+		options []string
+		wantErr bool
+	}{
+		{
+			name:    "basic session creation",
+			id:      "test_datagram_session",
+			options: nil,
+			wantErr: false,
+		},
+		{
+			name:    "session with options",
+			id:      "test_datagram_with_opts",
+			options: []string{"inbound.length=1", "outbound.length=1"},
+			wantErr: false,
+		},
+		{
+			name: "session with small tunnel config",
+			id:   "test_datagram_small",
+			options: []string{
+				"inbound.length=0",
+				"outbound.length=0",
+				"inbound.lengthVariance=0",
+				"outbound.lengthVariance=0",
+				"inbound.quantity=1",
+				"outbound.quantity=1",
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sam, keys := setupTestSAM(t)
+			defer sam.Close()
+
+			session, err := NewDatagramSession(sam, tt.id, keys, tt.options)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("NewDatagramSession() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			if err == nil {
+				// Verify session properties
+				if session.ID() != tt.id {
+					t.Errorf("Session ID = %v, want %v", session.ID(), tt.id)
+				}
+
+				if session.Keys().Addr().Base32() != keys.Addr().Base32() {
+					t.Error("Session keys don't match provided keys")
+				}
+
+				addr := session.Addr()
+				if addr.Base32() == "" {
+					t.Error("Session address is empty")
+				}
+
+				// Clean up
+				if err := session.Close(); err != nil {
+					t.Errorf("Failed to close session: %v", err)
+				}
+			}
+		})
+	}
+}
+
+func TestDatagramSession_Close(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_close", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+
+	// Close the session
+	err = session.Close()
+	if err != nil {
+		t.Errorf("Close() error = %v", err)
+	}
+
+	// Closing again should not error
+	err = session.Close()
+	if err != nil {
+		t.Errorf("Second Close() error = %v", err)
+	}
+}
+
+func TestDatagramSession_Addr(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_addr", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	addr := session.Addr()
+	expectedAddr := keys.Addr()
+
+	if addr.Base32() != expectedAddr.Base32() {
+		t.Errorf("Addr() = %v, want %v", addr.Base32(), expectedAddr.Base32())
+	}
+}
+
+func TestDatagramSession_NewReader(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_reader", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	reader := session.NewReader()
+	if reader == nil {
+		t.Error("NewReader() returned nil")
+	}
+
+	if reader.session != session {
+		t.Error("Reader session reference is incorrect")
+	}
+
+	// Verify channels are initialized
+	if reader.recvChan == nil {
+		t.Error("Reader recvChan is nil")
+	}
+	if reader.errorChan == nil {
+		t.Error("Reader errorChan is nil")
+	}
+	if reader.closeChan == nil {
+		t.Error("Reader closeChan is nil")
+	}
+}
+
+func TestDatagramSession_NewWriter(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_writer", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	writer := session.NewWriter()
+	if writer == nil {
+		t.Error("NewWriter() returned nil")
+	}
+
+	if writer.session != session {
+		t.Error("Writer session reference is incorrect")
+	}
+
+	if writer.timeout != 30 {
+		t.Errorf("Writer timeout = %v, want 30", writer.timeout)
+	}
+}
+
+func TestDatagramSession_PacketConn(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_packetconn", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	conn := session.PacketConn()
+	if conn == nil {
+		t.Error("PacketConn() returned nil")
+	}
+
+	datagramConn, ok := conn.(*DatagramConn)
+	if !ok {
+		t.Error("PacketConn() did not return a DatagramConn")
+	}
+
+	if datagramConn.session != session {
+		t.Error("DatagramConn session reference is incorrect")
+	}
+
+	if datagramConn.reader == nil {
+		t.Error("DatagramConn reader is nil")
+	}
+
+	if datagramConn.writer == nil {
+		t.Error("DatagramConn writer is nil")
+	}
+}
+
+func TestDatagramSession_ConcurrentOperations(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	session, err := NewDatagramSession(sam, "test_concurrent", keys, nil)
+	if err != nil {
+		t.Fatalf("Failed to create session: %v", err)
+	}
+	defer session.Close()
+
+	// Test concurrent reader creation
+	done := make(chan bool, 10)
+	for i := 0; i < 10; i++ {
+		go func() {
+			reader := session.NewReader()
+			if reader == nil {
+				t.Error("NewReader returned nil")
+			}
+			done <- true
+		}()
+	}
+
+	// Test concurrent writer creation
+	for i := 0; i < 10; i++ {
+		go func() {
+			writer := session.NewWriter()
+			if writer == nil {
+				t.Error("NewWriter returned nil")
+			}
+			done <- true
+		}()
+	}
+
+	// Wait for all goroutines
+	timeout := time.After(5 * time.Second)
+	for i := 0; i < 20; i++ {
+		select {
+		case <-done:
+			// OK
+		case <-timeout:
+			t.Fatal("Timeout waiting for concurrent operations")
+		}
+	}
+}
+
+func TestDatagramAddr_Network(t *testing.T) {
+	addr := &DatagramAddr{}
+	if addr.Network() != "i2p-datagram" {
+		t.Errorf("Network() = %v, want i2p-datagram", addr.Network())
+	}
+}
+
+func TestDatagramAddr_String(t *testing.T) {
+	if testing.Short() {
+		t.Skip("Skipping integration test in short mode")
+	}
+
+	sam, keys := setupTestSAM(t)
+	defer sam.Close()
+
+	addr := &DatagramAddr{addr: keys.Addr()}
+	expected := keys.Addr().Base32()
+
+	if addr.String() != expected {
+		t.Errorf("String() = %v, want %v", addr.String(), expected)
+	}
+}
--- a/datagram/types.go
+++ b/datagram/types.go
@ -0,0 +1,56 @@
+package datagram
+
+import (
+	"sync"
+	"time"
+
+	"github.com/go-i2p/go-sam-go/common"
+	"github.com/go-i2p/i2pkeys"
+)
+
+// DatagramSession represents a datagram session that can send and receive datagrams
+type DatagramSession struct {
+	*common.BaseSession
+	sam     *common.SAM
+	options []string
+	mu      sync.RWMutex
+	closed  bool
+}
+
+// DatagramReader handles incoming datagram reception
+type DatagramReader struct {
+	session   *DatagramSession
+	recvChan  chan *Datagram
+	errorChan chan error
+	closeChan chan struct{}
+	closed    bool
+	mu        sync.RWMutex
+}
+
+// DatagramWriter handles outgoing datagram transmission
+type DatagramWriter struct {
+	session *DatagramSession
+	timeout time.Duration
+}
+
+// Datagram represents an I2P datagram message
+type Datagram struct {
+	Data   []byte
+	Source i2pkeys.I2PAddr
+	Local  i2pkeys.I2PAddr
+}
+
+// DatagramAddr implements net.Addr for I2P datagram addresses
+type DatagramAddr struct {
+	addr i2pkeys.I2PAddr
+}
+
+// DatagramConn implements net.PacketConn for I2P datagrams
+type DatagramConn struct {
+	session    *DatagramSession
+	reader     *DatagramReader
+	writer     *DatagramWriter
+	remoteAddr *i2pkeys.I2PAddr
+	mu         sync.RWMutex
+	closed     bool
+}
--- a/datagram/types_test.go
+++ b/datagram/types_test.go
@ -0,0 +1,14 @@
+package datagram
+
+import (
+	"net"
+
+	"github.com/go-i2p/go-sam-go/common"
+)
+
+var (
+	ds  common.Session = &DatagramSession{}
+	dl  net.Listener   = &DatagramListener{}
+	dc  net.PacketConn = &DatagramConn{}
+	dcc net.Conn       = &DatagramConn{}
+)
--- a/datagram/write.go
+++ b/datagram/write.go
@ -0,0 +1,96 @@
+package datagram
+
+import (
+	"encoding/base64"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-i2p/i2pkeys"
+	"github.com/samber/oops"
+	"github.com/sirupsen/logrus"
+)
+
+// SetTimeout sets the timeout for datagram operations
+func (w *DatagramWriter) SetTimeout(timeout time.Duration) *DatagramWriter {
+	w.timeout = timeout
+	return w
+}
+
+// SendDatagram sends a datagram to the specified destination
+func (w *DatagramWriter) SendDatagram(data []byte, dest i2pkeys.I2PAddr) error {
+	w.session.mu.RLock()
+	if w.session.closed {
+		w.session.mu.RUnlock()
+		return oops.Errorf("session is closed")
+	}
+	w.session.mu.RUnlock()
+
+	logger := log.WithFields(logrus.Fields{
+		"session_id":  w.session.ID(),
+		"destination": dest.Base32(),
+		"size":        len(data),
+	})
+	logger.Debug("Sending datagram")
+
+	// Encode the data as base64
+	encodedData := base64.StdEncoding.EncodeToString(data)
+
+	// Create the DATAGRAM SEND command
+	sendCmd := fmt.Sprintf("DATAGRAM SEND ID=%s DESTINATION=%s SIZE=%d\n%s\n",
+		w.session.ID(), dest.Base64(), len(data), encodedData)
+
+	logger.WithField("command", strings.Split(sendCmd, "\n")[0]).Debug("Sending DATAGRAM SEND")
+
+	// Send the command
+	_, err := w.session.Write([]byte(sendCmd))
+	if err != nil {
+		logger.WithError(err).Error("Failed to send datagram")
+		return oops.Errorf("failed to send datagram: %w", err)
+	}
+
+	// Read the response
+	buf := make([]byte, 1024)
+	n, err := w.session.Read(buf)
+	if err != nil {
+		logger.WithError(err).Error("Failed to read send response")
+		return oops.Errorf("failed to read send response: %w", err)
+	}
+
+	response := string(buf[:n])
+	logger.WithField("response", response).Debug("Received send response")
+
+	// Parse the response
+	if err := w.parseSendResponse(response); err != nil {
+		return err
+	}
+
+	logger.Debug("Successfully sent datagram")
+	return nil
+}
+
+// parseSendResponse parses the DATAGRAM STATUS response
+func (w *DatagramWriter) parseSendResponse(response string) error {
+	if strings.Contains(response, "RESULT=OK") {
+		return nil
+	}
+
+	switch {
+	case strings.Contains(response, "RESULT=CANT_REACH_PEER"):
+		return oops.Errorf("cannot reach peer")
+	case strings.Contains(response, "RESULT=I2P_ERROR"):
+		return oops.Errorf("I2P internal error")
+	case strings.Contains(response, "RESULT=INVALID_KEY"):
+		return oops.Errorf("invalid destination key")
+	case strings.Contains(response, "RESULT=INVALID_ID"):
+		return oops.Errorf("invalid session ID")
+	case strings.Contains(response, "RESULT=TIMEOUT"):
+		return oops.Errorf("send timeout")
+	default:
+		if strings.HasPrefix(response, "DATAGRAM STATUS RESULT=") {
+			result := strings.TrimSpace(response[23:])
+			return oops.Errorf("send failed: %s", result)
+		}
+		return oops.Errorf("unexpected response format: %s", response)
+	}
+}
--- a/stream/README.md
+++ b/stream/README.md
@ -0,0 +1,23 @@
+# go-sam-go/stream
+
+High-level streaming library for reliable TCP-like connections over I2P using the SAMv3 protocol.
+
+## Installation
+
+Install using Go modules with the package path `github.com/go-i2p/go-sam-go/stream`.
+
+## Usage
+
+The package provides TCP-like streaming connections over I2P networks. [`StreamSession`](stream/types.go) manages the connection lifecycle, [`StreamListener`](stream/types.go) handles incoming connections, and [`StreamConn`](stream/types.go) implements the standard `net.Conn` interface for seamless integration with existing Go networking code.
+
+Create sessions using [`NewStreamSession`](stream/session.go), establish listeners with [`Listen()`](stream/session.go), and dial outbound connections using [`Dial()`](stream/session.go) or [`DialI2P()`](stream/session.go). The implementation supports context-based timeouts, concurrent operations, and automatic connection management.
+
+Key features include full `net.Listener` and `net.Conn` compatibility, I2P address resolution, configurable tunnel parameters, and comprehensive error handling with proper resource cleanup.
+
+## Dependencies
+
+- github.com/go-i2p/go-sam-go/common - Core SAM protocol implementation
+- github.com/go-i2p/i2pkeys - I2P cryptographic key handling  
+- github.com/go-i2p/logger - Logging functionality
+- github.com/sirupsen/logrus - Structured logging
+- github.com/samber/oops - Enhanced error handling