Websocket server

protocol/ws/v2/go.sum

@@ -34,6 +34,8 @@ github.com/hashicorp/golang-lru v0.5.3 h1:YPkqC67at8FYaadspW/6uE0COsBxS2656RLEr8
 github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/klauspost/compress v1.10.3 h1:OP96hzwJVBIHYU52pVTI6CczrxPvrGfgqF9N5eTO0Q8=
 github.com/klauspost/compress v1.10.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
@@ -43,7 +45,9 @@ github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgx
 github.com/lightstep/tracecontext.go v0.0.0-20181129014701-1757c391b1ac h1:+2b6iGRJe3hvV/yVXrd41yVEjxuFHxasJqDhkIjS4gk=
 github.com/lightstep/tracecontext.go v0.0.0-20181129014701-1757c391b1ac/go.mod h1:Frd2bnT3w5FB5q49ENTfVlztJES+1k/7lyWX2+9gq/M=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=

protocol/ws/v2/client_protocol.go → protocol/ws/v2/protocol.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net/http"
 	"sync"
 
 	"nhooyr.io/websocket"
@@ -16,11 +17,11 @@ import (
 	"github.com/cloudevents/sdk-go/v2/protocol"
 )
 
-// ClientProtocol implements protocol.Receiver, protocol.Sender and protocol.Closer.
+// Protocol implements protocol.Receiver, protocol.Sender and protocol.Closer.
 // Note: when you use client.StartReceiver with this protocol, you can use just one
 // goroutine to poll this protocol, because the protocol itself cannot handle multiple
 // received messages at same time (WS has no multiplexing!)
-type ClientProtocol struct {
+type Protocol struct {
 	conn *websocket.Conn
 
 	format      format.Format
@@ -30,8 +31,26 @@ type ClientProtocol struct {
 	connOwned    bool // whether this protocol created the connection
 }
 
-// Dial wraps websocket.Dial and creates the ClientProtocol.
-func Dial(ctx context.Context, u string, opts *websocket.DialOptions) (*ClientProtocol, error) {
+// Dial wraps websocket.Dial and creates the Protocol.
+func Accept(ctx context.Context, w http.ResponseWriter, r *http.Request, opts *websocket.AcceptOptions) (*Protocol, error) {
+	if opts == nil {
+		opts = &websocket.AcceptOptions{}
+	}
+	opts.Subprotocols = SupportedSubprotocols
+	c, err := websocket.Accept(w, r, opts)
+	if err != nil {
+		return nil, err
+	}
+	p, err := NewProtocol(c)
+	if err != nil {
+		return nil, err
+	}
+	p.connOwned = true
+	return p, nil
+}
+
+// Dial wraps websocket.Dial and creates the Protocol.
+func Dial(ctx context.Context, u string, opts *websocket.DialOptions) (*Protocol, error) {
 	if opts == nil {
 		opts = &websocket.DialOptions{}
 	}
@@ -40,38 +59,38 @@ func Dial(ctx context.Context, u string, opts *websocket.DialOptions) (*ClientPr
 	if err != nil {
 		return nil, err
 	}
-	p, err := NewClientProtocol(c)
+	p, err := NewProtocol(c)
 	if err != nil {
 		return nil, err
 	}
 	p.connOwned = true
 	return p, nil
 }
 
-// NewClientProtocol wraps a websocket.Conn in a type that implements protocol.Receiver, protocol.Sender and protocol.Closer.
-// Look at ClientProtocol for more details.
-func NewClientProtocol(c *websocket.Conn) (*ClientProtocol, error) {
+// NewProtocol wraps a websocket.Conn in a type that implements protocol.Receiver, protocol.Sender and protocol.Closer.
+// Look at Protocol for more details.
+func NewProtocol(c *websocket.Conn) (*Protocol, error) {
 	f, messageType, err := resolveFormat(c.Subprotocol())
 	if err != nil {
 		return nil, err
 	}
-	return &ClientProtocol{
+	return &Protocol{
 		conn:        c,
 		format:      f,
 		messageType: messageType,
 		connOwned:   false,
 	}, nil
 }
 
-func (c *ClientProtocol) Send(ctx context.Context, m binding.Message, transformers ...binding.Transformer) error {
+func (c *Protocol) Send(ctx context.Context, m binding.Message, transformers ...binding.Transformer) error {
 	writer, err := c.conn.Writer(ctx, c.messageType)
 	if err != nil {
 		return err
 	}
 	return utils.WriteStructured(ctx, m, writer, transformers...)
 }
 
-func (c *ClientProtocol) Receive(ctx context.Context) (binding.Message, error) {
+func (c *Protocol) Receive(ctx context.Context) (binding.Message, error) {
 	c.receiverLock.Lock()
 	m, err := c.UnsafeReceive(ctx)
 	if m != nil {
@@ -86,9 +105,9 @@ func (c *ClientProtocol) Receive(ctx context.Context) (binding.Message, error) {
 
 // UnsafeReceive is like Receive, except it doesn't guard from multiple invocations
 // from different goroutines.
-func (c *ClientProtocol) UnsafeReceive(ctx context.Context) (binding.Message, error) {
+func (c *Protocol) UnsafeReceive(ctx context.Context) (binding.Message, error) {
 	messageType, reader, err := c.conn.Reader(ctx)
-	if errors.Is(err, io.EOF) || errors.Is(err, websocket.CloseError{}) || (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
+	if errors.Is(err, io.EOF) || errors.As(err, &websocket.CloseError{}) || (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
 		return nil, io.EOF
 	}
 	if err != nil {
@@ -109,7 +128,7 @@ func consumeStream(reader io.Reader) {
 	ioutil.ReadAll(reader)
 }
 
-func (c *ClientProtocol) Close(ctx context.Context) error {
+func (c *Protocol) Close(ctx context.Context) error {
 	if c.connOwned {
 		statusCode := websocket.StatusNormalClosure
 		if val := ctx.Value(codeKey{}); val != nil {
@@ -126,6 +145,6 @@ func (c *ClientProtocol) Close(ctx context.Context) error {
 	return nil
 }
 
-var _ protocol.Receiver = (*ClientProtocol)(nil)
-var _ protocol.Sender = (*ClientProtocol)(nil)
-var _ protocol.Closer = (*ClientProtocol)(nil)
+var _ protocol.Receiver = (*Protocol)(nil)
+var _ protocol.Sender = (*Protocol)(nil)
+var _ protocol.Closer = (*Protocol)(nil)

protocol/ws/v2/client_protocol_test.go → protocol/ws/v2/protocol_test.go

@@ -74,6 +74,51 @@ func TestClientProtocolPingPongWithClient(t *testing.T) {
 	AssertEvent(t, pong, HasId("2"), HasType("pong"))
 }
 
+func TestClientServerProtocolPingPong(t *testing.T) {
+	server := httptest.NewServer(pingPongProtocolHandler(t))
+	defer server.Close()
+
+	p, err := Dial(context.TODO(), server.URL, nil)
+	require.NoError(t, err)
+
+	ping := pingEvent()
+	require.NoError(t, p.Send(context.TODO(), binding.ToMessage(&ping)))
+
+	receivedMessage, err := p.Receive(context.TODO())
+	require.NoError(t, err)
+
+	pong, err := binding.ToEvent(context.TODO(), receivedMessage)
+	require.NoError(t, err)
+
+	AssertEvent(t, *pong, HasId("2"), HasType("pong"))
+
+	require.NoError(t, p.Close(context.TODO()))
+}
+
+func pingPongProtocolHandler(t *testing.T) http.HandlerFunc {
+	return func(writer http.ResponseWriter, request *http.Request) {
+		ctx := request.Context()
+		p, err := Accept(ctx, writer, request, &websocket.AcceptOptions{Subprotocols: SupportedSubprotocols})
+
+		require.NoError(t, err)
+		require.Equal(t, JsonSubprotocol, p.conn.Subprotocol())
+
+		m, err := p.Receive(ctx)
+		require.NoError(t, err)
+
+		ping, err := binding.ToEvent(ctx, m)
+		require.NoError(t, err)
+		AssertEvent(t, *ping, HasId("1"), HasType("ping"))
+
+		pong := ping.Clone()
+		pong.SetID("2")
+		pong.SetType("pong")
+
+		require.NoError(t, p.Send(ctx, binding.ToMessage(&pong)))
+		require.NoError(t, p.Close(ctx))
+	}
+}
+
 func pingPongHandler(t *testing.T) http.HandlerFunc {
 	return func(writer http.ResponseWriter, request *http.Request) {
 		c, err := websocket.Accept(writer, request, &websocket.AcceptOptions{Subprotocols: SupportedSubprotocols})

samples/README.md

@@ -37,4 +37,5 @@ You can grab them and copy-paste in your project to start using sdk-go.
   * [Sender](./stan/sender): Receive events using the CloudEvents Client.
   * [Sender & Receiver](./stan/sender-receiver): Send and receive events using the same NATS client.
 * WebSockets
-  * [Client](./ws/client): Sends and receive events, from client side, using the CloudEvents Client.
+  * [Client](./ws/client): Sends and receive events, from client side, using the CloudEvents Client.
+  * [Server](./ws/server): Sends and receive events, from server side, using the CloudEvents WebSocket implementation.

samples/ws/go.sum

@@ -49,6 +49,8 @@ github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/klauspost/compress v1.10.3 h1:OP96hzwJVBIHYU52pVTI6CczrxPvrGfgqF9N5eTO0Q8=
 github.com/klauspost/compress v1.10.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=

samples/ws/server/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"log"
+	"net/http"
+
+	cews "github.com/cloudevents/sdk-go/protocol/ws/v2"
+	"github.com/cloudevents/sdk-go/v2/binding"
+)
+
+func main() {
+	ctx := context.Background()
+
+	err := http.ListenAndServe(":8080", http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		p, err := cews.Accept(ctx, writer, request, nil)
+		if err != nil {
+			log.Printf("Error while accepting a websocket request: %v\n", err)
+			writer.WriteHeader(http.StatusInternalServerError)
+		}
+		defer p.Close(ctx)
+
+		message, err := p.Receive(ctx)
+		if err != nil {
+			log.Printf("Error while receiving a websocket message: %v\n", err)
+			return
+		}
+
+		receivedEvent, err := binding.ToEvent(ctx, message)
+		if err != nil {
+			log.Printf("Error while parsing the message: %v\n", err)
+			return
+		}
+
+		log.Printf("Received event:\n%s", receivedEvent)
+
+		// Echo the event back
+		err = p.Send(ctx, binding.ToMessage(receivedEvent))
+		if err != nil {
+			log.Printf("Error while echoing the event back: %v\n", err)
+			return
+		}
+	}))
+	if err != nil {
+		log.Fatalf("failed to start the listener: %v\n", err)
+	}
+}