2024-05-05 12:44:09: merge [master:190abc0ed5fa8e683e1e03cbc7ca4ac0a4…

…7a0781];

README.md

@@ -56,6 +56,8 @@ func main() {
 		//you have access to original HTTP request
 		fmt.Printf("id: %d, url: %s, connected\n", conn.Id(), conn.HttpRequest().URL)
 		srv.Rooms().Join(conn, "goodClients") //leave automatically then disconnected
+
+		conn.Data().Set("key", "value") //put any data associative with connection
 	})
 
 	//callback to handle disconnection

client_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/txix-open/etp/v3"
 	"github.com/txix-open/etp/v3/msg"
+	"github.com/txix-open/etp/v3/store"
 	"nhooyr.io/websocket"
 )
 
@@ -69,8 +70,14 @@ func TestClientHeavyConcurrency(t *testing.T) {
 		srv.Rooms().Join(conn, "connections")
 		err := conn.Emit(context.Background(), "hello", nil)
 		require.NoError(err)
+
+		conn.Data().Set("key", conn.Id())
 	}).On("closeMe", etp.HandlerFunc(func(ctx context.Context, conn *etp.Conn, event msg.Event) []byte {
-		err := conn.Close()
+		connId, err := store.Get[uint64](conn.Data(), "key")
+		require.NoError(err)
+		require.Equal(conn.Id(), connId)
+
+		err = conn.Close()
 		require.NoError(err)
 		return nil
 	})).OnDisconnect(srvHandler.OnDisconnect)

conn.go

@@ -8,13 +8,15 @@ import (
 	"github.com/txix-open/etp/v3/bpool"
 	"github.com/txix-open/etp/v3/internal"
 	"github.com/txix-open/etp/v3/msg"
+	"github.com/txix-open/etp/v3/store"
 	"nhooyr.io/websocket"
 )
 
 type Conn struct {
 	id      uint64
 	request *http.Request
 	ws      *websocket.Conn
+	data    *store.Store
 	acks    *internal.Acks
 }
 
@@ -27,6 +29,7 @@ func newConn(
 		id:      id,
 		request: request,
 		ws:      ws,
+		data:    store.New(),
 		acks:    internal.NewAcks(),
 	}
 }
@@ -39,6 +42,10 @@ func (c *Conn) HttpRequest() *http.Request {
 	return c.request
 }
 
+func (c *Conn) Data() *store.Store {
+	return c.data
+}
+
 func (c *Conn) Emit(ctx context.Context, event string, data []byte) error {
 	message := msg.Event{
 		Name:  event,

example/example_test.go

@@ -20,6 +20,8 @@ func ExampleServer() {
 		//you have access to original HTTP request
 		fmt.Printf("id: %d, url: %s, connected\n", conn.Id(), conn.HttpRequest().URL)
 		srv.Rooms().Join(conn, "goodClients") //leave automatically then disconnected
+
+		conn.Data().Set("key", "value") //put any data associative with connection
 	})
 
 	//callback to handle disconnection

go.mod

@@ -2,9 +2,7 @@ module github.com/txix-open/etp/v3
 
 go 1.22
 
-require (
-	nhooyr.io/websocket v1.8.11
-)
+require nhooyr.io/websocket v1.8.11
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect

store/store.go

@@ -0,0 +1,55 @@
+package store
+
+import (
+	"fmt"
+	"sync"
+)
+
+type Store struct {
+	data map[string]any
+	lock sync.Locker
+}
+
+func New() *Store {
+	return &Store{
+		data: make(map[string]any),
+		lock: &sync.Mutex{},
+	}
+}
+
+func (s *Store) Set(key string, value any) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.data[key] = value
+}
+
+func (s *Store) Range(f func(key string, value any) bool) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for k, v := range s.data {
+		if !f(k, v) {
+			return
+		}
+	}
+}
+
+func Get[T any](store *Store, key string) (T, error) {
+	store.lock.Lock()
+	defer store.lock.Unlock()
+
+	var empty T
+
+	value, ok := store.data[key]
+	if !ok {
+		return empty, fmt.Errorf("value for key %s not found", key)
+	}
+
+	casted, ok := value.(T)
+	if !ok {
+		return empty, fmt.Errorf("expected type %T, but got %T", empty, value)
+	}
+
+	return casted, nil
+}