Merge pull request #3 from spiral/feature/do-not-close-channel

bug(channel): Do not close closed channel

.github/dependabot.yml

@@ -0,0 +1,12 @@
+# To get started with Dependabot version updates, you'll need to specify which
+# package ecosystems to update and where the package manifests are located.
+# Please see the documentation for all configuration options:
+# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+  - package-ecosystem: gomod # See documentation for possible values
+    directory: "/" # Location of package manifests
+    schedule:
+      interval: daily
+

.github/workflows/linux.yml

@@ -0,0 +1,85 @@
+name: Linux
+
+on:
+  push:
+  pull_request:
+    branches:
+      # Branches from forks have the form 'user:branch-name' so we only run
+      # this job on pull_request events for branches that look like fork
+      # branches. Without this we would end up running this job twice for non
+      # forked PRs, once for the push and then once for opening the PR.
+      - "**:**"
+
+jobs:
+  golang:
+    name: Build (Go ${{ matrix.go }}, PHP ${{ matrix.php }}, OS ${{matrix.os}})
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 60
+    strategy:
+      fail-fast: false
+      matrix:
+        php: [ "7.2","7.3", "7.4", "8.0" ]
+        go: [ "1.14", "1.15" ]
+        os: [ ubuntu-20.04 ]
+    steps:
+      - name: Set up Go ${{ matrix.go }}
+        uses: actions/setup-go@v2 # action page: <https://github.com/actions/setup-go>
+        with:
+          go-version: ${{ matrix.go }}
+
+      - name: Set up PHP ${{ matrix.php }}
+        uses: shivammathur/setup-php@v2 # action page: <https://github.com/shivammathur/setup-php>
+        with:
+          php-version: ${{ matrix.php }}
+          extensions: sockets
+
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Get Composer Cache Directory
+        id: composer-cache
+        run: echo "::set-output name=dir::$(composer config cache-files-dir)"
+
+      - name: Init Composer Cache # Docs: <https://git.io/JfAKn#php---composer>
+        uses: actions/cache@v2
+        with:
+          path: ${{ steps.composer-cache.outputs.dir }}
+          key: ${{ runner.os }}-composer-${{ matrix.php }}-${{ hashFiles('**/composer.json') }}
+          restore-keys: ${{ runner.os }}-composer-
+
+      - name: Install Composer dependencies
+        run: composer update --prefer-dist --no-progress --ansi
+
+      - name: Init Go modules Cache # Docs: <https://git.io/JfAKn#go---modules>
+        uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ runner.os }}-go-
+
+      - name: Install Go dependencies
+        run: go mod download
+
+      - name: Run golang tests on Linux
+        run: |
+          mkdir ./coverage-ci
+          go test -v -race -cover -coverpkg=./... -coverprofile=./coverage-ci/broadcast-ws.txt -covermode=atomic
+
+      - uses: codecov/codecov-action@v1 # Docs: <https://github.com/codecov/codecov-action>
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          file: ./coverage-ci/broadcast-ws.txt
+          fail_ci_if_error: false
+
+  golangci-lint:
+    name: Golang-CI (lint)
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Run linter
+        uses: golangci/golangci-lint-action@v2 # Action page: <https://github.com/golangci/golangci-lint-action>
+        with:
+          version: v1.35 # without patch version
+          only-new-issues: false # show only new issues if it's a pull request

.github/workflows/macos.yml

@@ -0,0 +1,66 @@
+name: macOS
+
+on:
+  push:
+  pull_request:
+    branches:
+      # Branches from forks have the form 'user:branch-name' so we only run
+      # this job on pull_request events for branches that look like fork
+      # branches. Without this we would end up running this job twice for non
+      # forked PRs, once for the push and then once for opening the PR.
+      - "**:**"
+
+jobs:
+  golang:
+    name: Build (Go ${{ matrix.go }}, PHP ${{ matrix.php }}, OS ${{matrix.os}})
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 60
+    strategy:
+      fail-fast: false
+      matrix:
+        php: [ "7.2","7.3", "7.4", "8.0" ]
+        go: [ "1.14", "1.15" ]
+        os: [ macos-latest ]
+    steps:
+      - name: Set up Go ${{ matrix.go }}
+        uses: actions/setup-go@v2 # action page: <https://github.com/actions/setup-go>
+        with:
+          go-version: ${{ matrix.go }}
+
+      - name: Set up PHP ${{ matrix.php }}
+        uses: shivammathur/setup-php@v2 # action page: <https://github.com/shivammathur/setup-php>
+        with:
+          php-version: ${{ matrix.php }}
+          extensions: sockets
+
+      - name: Check out code
+        uses: actions/checkout@v2
+
+      - name: Get Composer Cache Directory
+        id: composer-cache
+        run: echo "::set-output name=dir::$(composer config cache-files-dir)"
+
+      - name: Init Composer Cache # Docs: <https://git.io/JfAKn#php---composer>
+        uses: actions/cache@v2
+        with:
+          path: ${{ steps.composer-cache.outputs.dir }}
+          key: ${{ runner.os }}-composer-${{ matrix.php }}-${{ hashFiles('**/composer.json') }}
+          restore-keys: ${{ runner.os }}-composer-
+
+      - name: Install Composer dependencies
+        run: composer update --prefer-dist --no-progress --ansi
+
+      - name: Init Go modules Cache # Docs: <https://git.io/JfAKn#go---modules>
+        uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ runner.os }}-go-
+
+      - name: Install Go dependencies
+        run: go mod download
+
+      - name: Run golang tests
+        run: |
+          mkdir ./coverage-ci
+          go test -v -race -cover -coverpkg=./... -coverprofile=./coverage-ci/broadcast-ws.txt -covermode=atomic

.gitignore

@@ -1,6 +1,6 @@
 .idea
 go.sum
-vendor
+vendor_php
 composer.lock
 .php_cs.cache
-tests/log.txt
+tests/log.txt

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 {
@@ -31,7 +32,7 @@ func (w *accessValidator) copy(rw http.ResponseWriter) {
 		}
 	}
 
-	io.Copy(rw, w.buffer)
+	_, _ = io.Copy(rw, w.buffer)
 }
 
 // Header returns the header map that will be sent by WriteHeader.

access_validator_test.go

@@ -1,13 +1,14 @@
 package ws
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestResponseWrapper_Body(t *testing.T) {
 	w := newValidator()
-	w.Write([]byte("hello"))
+	_, _ =w.Write([]byte("hello"))
 
 	assert.Equal(t, []byte("hello"), w.Body())
 }

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() {

composer.json

@@ -9,6 +9,9 @@
       "email": "[email protected]"
     }
   ],
+  "config": {
+    "vendor-dir": "vendor_php"
+  },
   "require": {
     "php": ">=7.2",
     "ext-json": "*",

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
@@ -71,7 +72,6 @@ func (cp *connPool) disconnect(conn *websocket.Conn) error {
 	}
 
 	delete(cp.conns, conn)
-	close(ctx.upstream)
 
 	return conn.Close()
 }
@@ -117,7 +117,7 @@ func (cp *connPool) close() {
 		}
 
 		delete(cp.conns, conn)
-		close(ctx.upstream)
+
 		if err := conn.Close(); err != nil {
 			cp.errHandler(err, conn)
 		}

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/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/gorilla/websocket v1.4.2
+	github.com/prometheus/client_golang v1.7.1
+	github.com/sirupsen/logrus v1.6.0
+	github.com/spf13/cobra v1.0.0
+	github.com/spiral/broadcast/v2 v2.0.5
+	github.com/spiral/roadrunner v1.9.2
+	github.com/stretchr/testify v1.6.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.