Move close channel routines after connection closed

access_validator.go

@@ -2,10 +2,11 @@ package ws
 
 import (
 	"bytes"
-	"github.com/spiral/roadrunner/service/http/attributes"
 	"io"
 	"net/http"
 	"strings"
+
+	"github.com/spiral/roadrunner/service/http/attributes"
 )
 
 type accessValidator struct {

access_validator_test.go

@@ -1,8 +1,9 @@
 package ws
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestResponseWrapper_Body(t *testing.T) {

cmd/rr-ws/ws/debug.go

@@ -1,13 +1,14 @@
 package ws
 
 import (
+	"strings"
+
 	"github.com/gorilla/websocket"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	ws "github.com/spiral/broadcast-ws"
 	rr "github.com/spiral/roadrunner/cmd/rr/cmd"
 	"github.com/spiral/roadrunner/cmd/util"
-	"strings"
 )
 
 func init() {

config_test.go

@@ -2,9 +2,10 @@ package ws
 
 import (
 	"encoding/json"
+	"testing"
+
 	"github.com/spiral/roadrunner/service"
 	"github.com/stretchr/testify/assert"
-	"testing"
 )
 
 type mockCfg struct{ cfg string }

conn_context.go

@@ -2,8 +2,9 @@ package ws
 
 import (
 	"encoding/json"
+
 	"github.com/gorilla/websocket"
-	"github.com/spiral/broadcast"
+	"github.com/spiral/broadcast/v2"
 )
 
 // ConnContext carries information about websocket connection and it's topics.

conn_context_test.go

@@ -1,8 +1,9 @@
 package ws
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestConnContext_ManageTopics(t *testing.T) {

conn_pool.go

@@ -2,9 +2,10 @@ package ws
 
 import (
 	"errors"
-	"github.com/gorilla/websocket"
-	"github.com/spiral/broadcast"
 	"sync"
+
+	"github.com/gorilla/websocket"
+	"github.com/spiral/broadcast/v2"
 )
 
 // manages a set of websocket connections
@@ -116,10 +117,11 @@ func (cp *connPool) close() {
 			cp.errHandler(err, conn)
 		}
 
-		delete(cp.conns, conn)
-		close(ctx.upstream)
 		if err := conn.Close(); err != nil {
 			cp.errHandler(err, conn)
 		}
+
+		delete(cp.conns, conn)
+		close(ctx.upstream)
 	}
 }

go.mod

@@ -1,13 +1,13 @@
 module github.com/spiral/broadcast-ws
 
-go 1.12
+go 1.15
 
 require (
 	github.com/gorilla/websocket v1.4.1
-	github.com/prometheus/client_golang v1.2.1 // indirect
+	github.com/prometheus/client_golang v1.5.0
 	github.com/sirupsen/logrus v1.4.2
-	github.com/spf13/cobra v0.0.5 // indirect
-	github.com/spiral/broadcast v0.0.0-20191206140608-766959683e74
-	github.com/spiral/roadrunner v1.5.2
-	github.com/stretchr/testify v1.3.0
+	github.com/spf13/cobra v0.0.6
+	github.com/spiral/broadcast/v2 v2.0.5-beta1
+	github.com/spiral/roadrunner v1.6.4
+	github.com/stretchr/testify v1.5.1
 )

service.go

@@ -2,14 +2,15 @@ package ws
 
 import (
 	"encoding/json"
+	"net/http"
+	"sync"
+	"sync/atomic"
+
 	"github.com/gorilla/websocket"
-	"github.com/spiral/broadcast"
+	"github.com/spiral/broadcast/v2"
 	"github.com/spiral/roadrunner/service/env"
 	rhttp "github.com/spiral/roadrunner/service/http"
 	"github.com/spiral/roadrunner/service/rpc"
-	"net/http"
-	"sync"
-	"sync/atomic"
 )
 
 // ID defines service id.

service_test.go

@@ -2,21 +2,22 @@ package ws
 
 import (
 	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
 	"github.com/gorilla/websocket"
 	"github.com/sirupsen/logrus"
 	"github.com/sirupsen/logrus/hooks/test"
-	"github.com/spiral/broadcast"
+	"github.com/spiral/broadcast/v2"
 	"github.com/spiral/roadrunner/service"
 	"github.com/spiral/roadrunner/service/env"
 	rrhttp "github.com/spiral/roadrunner/service/http"
 	"github.com/spiral/roadrunner/service/rpc"
 	"github.com/stretchr/testify/assert"
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"strings"
-	"testing"
-	"time"
 )
 
 type testCfg struct {