Upgrade to AnyCable 1.4

.github/workflows/lint.yml

@@ -10,8 +10,12 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - uses: actions/setup-go@v2
       with:
-        go-version: 1.19.x
-    - run: make lint
+        go-version: 1.20.x
+    - name: golangci-lint
+      uses: golangci/golangci-lint-action@v3
+      with:
+        version: v1.54.2
+        args: --timeout 3m

.github/workflows/test.yml

@@ -10,10 +10,10 @@ jobs:
   test:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
     - uses: actions/setup-go@v2
       with:
-        go-version: 1.19.x
+        go-version: 1.20.x
     - name: Run tests
       run: |
         make test

Makefile

@@ -40,7 +40,7 @@ test:
 
 bin/golangci-lint:
 	@test -x $$(go env GOPATH)/bin/golangci-lint || \
-		curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $$(go env GOPATH)/bin v1.48.0
+		curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $$(go env GOPATH)/bin v1.54.2
 
 lint: bin/golangci-lint
 	$$(go env GOPATH)/bin/golangci-lint run

README.md

@@ -10,7 +10,7 @@ Why building a WebSocket application with AnyCable-Go (and not other Go librarie
 
 Read more about [AnyCable-Go][anycable-docs].
 
-### Examples
+## Examples
 
 - [AnyCable + Twilio Streams](https://github.com/anycable/anycable-twilio-hanami-demo)
 
@@ -28,7 +28,7 @@ Make sure everything is ready by running `make lint` and `make test`.
 
 ## Development
 
-**NOTE:** Make sure Go 1.19+ installed.
+**NOTE:** Make sure Go 1.20+ installed.
 
 The following commands are available:
 

go.mod

@@ -1,9 +1,9 @@
 module github.com/anycable/mycable
 
-go 1.19
+go 1.20
 
 require (
-	github.com/anycable/anycable-go v1.3.0
+	github.com/anycable/anycable-go v1.4.5
 	github.com/apex/log v1.9.0
 	github.com/gorilla/websocket v1.5.0
 	github.com/urfave/cli/v2 v2.24.1
@@ -12,11 +12,14 @@ require (
 require (
 	github.com/FZambia/sentinel v1.1.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
+	github.com/fullstorydev/grpchan v1.0.2-0.20210924023933-8d2633e44524 // indirect
 	github.com/go-chi/chi/v5 v5.0.8 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/gomodule/redigo v1.8.8 // indirect
-	github.com/google/gops v0.3.23 // indirect
+	github.com/google/gops v0.3.27 // indirect
+	github.com/hofstadter-io/cinful v1.0.0 // indirect
+	github.com/jhump/protoreflect v1.5.0 // indirect
 	github.com/joomcode/errorx v1.1.0 // indirect
 	github.com/klauspost/compress v1.16.0 // indirect
 	github.com/matoous/go-nanoid v1.5.0 // indirect
@@ -29,16 +32,19 @@ require (
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/posthog/posthog-go v0.0.0-20221221115252-24dfed35d71a // indirect
+	github.com/redis/rueidis v1.0.17 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/smira/go-statsd v1.3.2 // indirect
-	github.com/stretchr/testify v1.8.1 // indirect
-	github.com/syossan27/tebata v0.0.0-20180602121909-b283fe4bc5ba // indirect
+	github.com/sony/gobreaker v0.5.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
+	github.com/xtgo/uuid v0.0.0-20140804021211-a0b114877d4c // indirect
 	go.uber.org/automaxprocs v1.5.1 // indirect
-	golang.org/x/crypto v0.6.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
-	golang.org/x/text v0.7.0 // indirect
+	golang.org/x/crypto v0.11.0 // indirect
+	golang.org/x/exp v0.0.0-20230626212559-97b1e661b5df // indirect
+	golang.org/x/net v0.12.0 // indirect
+	golang.org/x/sys v0.10.0 // indirect
+	golang.org/x/text v0.11.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	google.golang.org/genproto v0.0.0-20230216225411-c8e22ba71e44 // indirect
 	google.golang.org/grpc v1.53.0 // indirect

internal/fake_rpc/controller.go

@@ -35,6 +35,8 @@ func (c *Controller) Shutdown() error {
 }
 
 func (c *Controller) Authenticate(sid string, env *common.SessionEnv) (*common.ConnectResult, error) {
+	c.log.WithField("sid", sid).Debug("> Connected")
+
 	return &common.ConnectResult{
 		Status:        common.SUCCESS,
 		Identifier:    sid,
@@ -43,6 +45,8 @@ func (c *Controller) Authenticate(sid string, env *common.SessionEnv) (*common.C
 }
 
 func (c *Controller) Subscribe(sid string, env *common.SessionEnv, identifiers string, channel string) (*common.CommandResult, error) {
+	c.log.WithField("sid", sid).Debugf("> Subscribed to %s", channel)
+
 	res := &common.CommandResult{
 		Status:        common.SUCCESS,
 		Transmissions: []string{cableMessage("confirm_subscription", channel)},
@@ -51,6 +55,8 @@ func (c *Controller) Subscribe(sid string, env *common.SessionEnv, identifiers s
 }
 
 func (c *Controller) Unsubscribe(sid string, env *common.SessionEnv, identifiers string, channel string) (*common.CommandResult, error) {
+	c.log.WithField("sid", sid).Debugf("> Unubscribed from %s", channel)
+
 	res := &common.CommandResult{
 		Status: common.SUCCESS,
 	}
@@ -66,7 +72,7 @@ func (c *Controller) Perform(sid string, env *common.SessionEnv, id string, chan
 
 	action := payload["action"].(string)
 
-	c.log.Debugf("Perform action: %s, data: %v", action, payload)
+	c.log.WithField("sid", sid).Debugf("> Perform action: %s, data: %v", action, payload)
 
 	nextState := make(map[string]string)
 
@@ -83,6 +89,8 @@ func (c *Controller) Perform(sid string, env *common.SessionEnv, id string, chan
 }
 
 func (c *Controller) Disconnect(sid string, env *common.SessionEnv, id string, subscriptions []string) error {
+	c.log.WithField("sid", sid).Debug("> Disconnected")
+
 	return nil
 }
 

pkg/cli/cli.go

@@ -7,7 +7,7 @@ import (
 	aconfig "github.com/anycable/anycable-go/config"
 	"github.com/anycable/anycable-go/metrics"
 	"github.com/anycable/anycable-go/node"
-	"github.com/anycable/anycable-go/pubsub"
+	"github.com/anycable/anycable-go/server"
 	"github.com/anycable/anycable-go/ws"
 	"github.com/apex/log"
 	"github.com/gorilla/websocket"
@@ -30,20 +30,13 @@ func Run(conf *config.Config, anyconf *aconfig.Config) error {
 	return anycable.Run()
 }
 
-// NoopSubscriber is used to stub AnyCable pub/sub functionality
-type NoopSubscriber struct{}
-
-var _ pubsub.Subscriber = (*NoopSubscriber)(nil)
-
-func (NoopSubscriber) Start(done chan error) (err error) { return }
-func (NoopSubscriber) Shutdown() (err error)             { return }
-
 func initAnyCableRunner(appConf *config.Config, anyConf *aconfig.Config) (*acli.Runner, error) {
 	opts := []acli.Option{
 		acli.WithName("AnyCable"),
-		acli.WithSubscriber(func(h pubsub.Handler, c *aconfig.Config) (pubsub.Subscriber, error) {
-			return &NoopSubscriber{}, nil
-		}),
+		acli.WithDefaultSubscriber(),
+		acli.WithDefaultBroker(),
+		// Enable broadcasting
+		// acli.WithDefaultBroadcaster(),
 		acli.WithWebSocketEndpoint("/ws", myWebsocketHandler(appConf)),
 	}
 
@@ -60,17 +53,18 @@ func initAnyCableRunner(appConf *config.Config, anyConf *aconfig.Config) (*acli.
 
 func myWebsocketHandler(config *config.Config) func(n *node.Node, c *aconfig.Config) (http.Handler, error) {
 	return func(n *node.Node, c *aconfig.Config) (http.Handler, error) {
-		extractor := ws.DefaultHeadersExtractor{Headers: c.Headers, Cookies: c.Cookies}
+		extractor := server.DefaultHeadersExtractor{Headers: c.Headers, Cookies: c.Cookies}
 
 		executor := custom.NewExecutor(n)
 
 		log.WithField("context", "main").Infof("Handle custom WebSocket connections at ws://%s:%d/ws", c.Host, c.Port)
 
-		return ws.WebsocketHandler([]string{}, &extractor, &c.WS, func(wsc *websocket.Conn, info *ws.RequestInfo, callback func()) error {
+		return ws.WebsocketHandler([]string{}, &extractor, &c.WS, func(wsc *websocket.Conn, info *server.RequestInfo, callback func()) error {
 			wrappedConn := ws.NewConnection(wsc)
-			session := node.NewSession(n, wrappedConn, info.URL, info.Headers, info.UID)
-			session.SetEncoder(custom.Encoder{})
-			session.SetExecutor(executor)
+			session := node.NewSession(
+				n, wrappedConn, info.URL, info.Headers, info.UID,
+				node.WithEncoder(custom.Encoder{}), node.WithExecutor(executor),
+			)
 
 			_, err := n.Authenticate(session)
 

pkg/version/version.go

@@ -7,7 +7,7 @@ var (
 
 func init() {
 	if version == "" {
-		version = "0.1.0"
+		version = "0.2.0"
 	}
 
 	if sha != "" {