Merge pull request #4 from xconnio/client

add WAMP client/session implementation

client.go

@@ -0,0 +1,34 @@
+package xconn
+
+import (
+	"context"
+	"time"
+
+	"github.com/xconnio/wampproto-go/auth"
+)
+
+type Client struct {
+	Authenticator  auth.ClientAuthenticator
+	SerializerSpec WSSerializerSpec
+
+	DialTimeout time.Duration
+}
+
+func (c *Client) Connect(ctx context.Context, url string, realm string) (*Session, error) {
+	if c.SerializerSpec == nil {
+		c.SerializerSpec = JSONSerializerSpec
+	}
+
+	joiner := &WebSocketJoiner{
+		Authenticator:  c.Authenticator,
+		SerializerSpec: c.SerializerSpec,
+		DialTimeout:    c.DialTimeout,
+	}
+
+	base, err := joiner.Join(ctx, url, realm)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewSession(base, c.SerializerSpec.Serializer()), nil // nolint: contextcheck
+}

client_test.go

@@ -0,0 +1,90 @@
+package xconn_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/xconnio/xconn-go"
+)
+
+func connect(t *testing.T) *xconn.Session {
+	listener := startRouter(t, "realm1")
+	defer func() { _ = listener.Close() }()
+	address := fmt.Sprintf("ws://%s/ws", listener.Addr().String())
+
+	client := &xconn.Client{}
+
+	session, err := client.Connect(context.Background(), address, "realm1")
+	require.NoError(t, err)
+	require.NotNil(t, session)
+
+	return session
+}
+
+func TestCall(t *testing.T) {
+	session := connect(t)
+	t.Run("CallNoProc", func(t *testing.T) {
+		result, err := session.Call(context.Background(), "foo.bar", nil, nil, nil)
+		require.Error(t, err)
+		require.Nil(t, result)
+
+		var er *xconn.Error
+		errors.As(err, &er)
+		require.Equal(t, "wamp.error.no_such_procedure", er.URI)
+	})
+}
+
+func TestRegisterCall(t *testing.T) {
+	session := connect(t)
+	reg, err := session.Register(
+		context.Background(),
+		"foo.bar",
+		func(ctx context.Context, invocation *xconn.Invocation) (*xconn.Result, error) {
+			return &xconn.Result{Args: []any{"hello"}}, nil
+		},
+		nil,
+	)
+
+	require.NoError(t, err)
+	require.NotNil(t, reg)
+
+	t.Run("Call", func(t *testing.T) {
+		result, err := session.Call(context.Background(), "foo.bar", nil, nil, nil)
+		require.NoError(t, err)
+		require.NotNil(t, result)
+
+		require.Equal(t, "hello", result.Args[0])
+	})
+}
+
+func TestPublishSubscribe(t *testing.T) {
+	session := connect(t)
+	event1 := make(chan *xconn.Event, 1)
+	reg, err := session.Subscribe(
+		context.Background(),
+		"foo.bar",
+		func(event *xconn.Event) {
+			event1 <- event
+		},
+		nil,
+	)
+
+	require.NoError(t, err)
+	require.NotNil(t, reg)
+
+	t.Run("Publish", func(t *testing.T) {
+		opt := map[string]any{
+			"exclude_me": false,
+		}
+		err := session.Publish(context.Background(), "foo.bar", nil, nil, opt)
+		require.NoError(t, err)
+
+		event := <-event1
+		log.Println(event)
+	})
+}

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/gammazero/nexus/v3 v3.2.2
 	github.com/gobwas/ws v1.4.0
 	github.com/stretchr/testify v1.8.4
-	github.com/xconnio/wampproto-go v0.0.0-20240529204913-b111a4553cc8
+	github.com/xconnio/wampproto-go v0.0.0-20240530132134-a9a2ca11944a
 )
 
 require (

go.sum

@@ -24,8 +24,8 @@ github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAh
 github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
-github.com/xconnio/wampproto-go v0.0.0-20240529204913-b111a4553cc8 h1:IkY83ySxeDn5k2C6zBZ5I7Z8+VNFpseSPLv3enadLdI=
-github.com/xconnio/wampproto-go v0.0.0-20240529204913-b111a4553cc8/go.mod h1:BH0AFRLJ9POvVfxsFd9GyvA15U9o0XYQfq8TdkqO2vQ=
+github.com/xconnio/wampproto-go v0.0.0-20240530132134-a9a2ca11944a h1:0Vb6+/sNho0zn7kjRsdBvdGylFkz9uZZzeT1RrgKHIU=
+github.com/xconnio/wampproto-go v0.0.0-20240530132134-a9a2ca11944a/go.mod h1:BH0AFRLJ9POvVfxsFd9GyvA15U9o0XYQfq8TdkqO2vQ=
 go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
 golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=

joiner.go

@@ -95,7 +95,7 @@ func Join(cl Peer, realm string, serializer serializers.Serializer,
 			return nil, err
 		}
 
-		// nothing to send, this means the session was established.
+		// nothing to send, this means the proto was established.
 		if toSend == nil {
 			details, err := j.SessionDetails()
 			if err != nil {

joiner_test.go

@@ -34,8 +34,8 @@ func startRouter(t *testing.T, realm string) net.Listener {
 func TestJoin(t *testing.T) {
 	listener := startRouter(t, "realm1")
 	defer func() { _ = listener.Close() }()
-
 	address := fmt.Sprintf("ws://%s/ws", listener.Addr().String())
+
 	var joiner xconn.WebSocketJoiner
 	base, err := joiner.Join(context.Background(), address, "realm1")
 	require.NoError(t, err)

peer.go

@@ -47,6 +47,10 @@ func (b *baseSession) Write(payload []byte) error {
 	return b.client.Write(payload)
 }
 
+func (b *baseSession) Close() error {
+	return b.client.NetConn().Close()
+}
+
 func NewWebSocketPeer(conn net.Conn, protocol string, binary, server bool) (Peer, error) {
 	var wsReader ReaderFunc
 	var wsWriter WriterFunc