diff --git a/.github/workflows/windows.yml b/.github/workflows/windows.yml deleted file mode 100644 index f23f9b5d5..000000000 --- a/.github/workflows/windows.yml +++ /dev/null @@ -1,96 +0,0 @@ -name: Windows - -on: - push: - branches: - - master - - beta - - stable - tags-ignore: - - '**' - paths-ignore: - - '**.md' - pull_request: - paths-ignore: - - '**.md' - -jobs: - golang: - name: Build (Go ${{ matrix.go }}, PHP ${{ matrix.php }}, OS ${{matrix.os}}) - runs-on: ${{ matrix.os }} - timeout-minutes: 60 - strategy: - fail-fast: true - matrix: - php: [ "8.0" ] - go: [ "1.16" ] - os: [ windows-latest ] - steps: - - name: Set up Go ${{ matrix.go }} - uses: actions/setup-go@v2 # action page: - with: - go-version: ${{ matrix.go }} - - - name: Set up PHP ${{ matrix.php }} - uses: shivammathur/setup-php@v2 # action page: - 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: - 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: cd tests && composer update --prefer-dist --no-progress --ansi - - - name: Init Go modules Cache # Docs: - 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 Windows - run: | - docker-compose -f ./tests/docker-compose.yaml up -d - go test -v -race ./pkg/transport/pipe - go test -v -race ./pkg/transport/socket - go test -v -race ./pkg/pool - go test -v -race ./pkg/worker - go test -v -race ./pkg/bst - go test -v -race ./pkg/worker_watcher - go test -v -race ./plugins/http/config - go test -v -race ./plugins/server - go test -v -race ./tests/plugins/http - go test -v -race ./tests/plugins/informer - go test -v -race ./tests/plugins/reload - go test -v -race ./tests/plugins/server - go test -v -race ./tests/plugins/service - go test -v -race ./tests/plugins/status - go test -v -race ./tests/plugins/config - go test -v -race ./tests/plugins/gzip - go test -v -race ./tests/plugins/headers - go test -v -race ./tests/plugins/logger - go test -v -race ./tests/plugins/metrics - go test -v -race ./tests/plugins/redis - go test -v -race ./tests/plugins/resetter - go test -v -race ./tests/plugins/rpc - go test -v -race ./tests/plugins/kv - go test -v -race ./tests/plugins/broadcast - go test -v -race ./tests/plugins/websockets - go test -v -race ./plugins/websockets - docker-compose -f ./tests/docker-compose.yaml down diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f8e1733c..ea55a10d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ v2.3.1 (_.06.2021) ## 👀 New: - ✏️ Rework `broadcast` plugin. Add architecture diagrams to the `doc` folder. [PR](https://github.com/spiral/roadrunner/pull/732) +- ✏️ Add `Clear` method to the KV plugin RPC. [PR](https://github.com/spiral/roadrunner/pull/736) ## 🩹 Fixes: diff --git a/README.md b/README.md index 08580f860..ea100f246 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,6 @@ - diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go index 4b6752718..47d37cc24 100644 --- a/plugins/kv/drivers/boltdb/driver.go +++ b/plugins/kv/drivers/boltdb/driver.go @@ -18,6 +18,7 @@ import ( ) type Driver struct { + clearMu sync.RWMutex // db instance DB *bolt.DB // name should be UTF-8 @@ -373,6 +374,35 @@ func (d *Driver) TTL(keys ...string) (map[string]string, error) { return m, nil } +func (d *Driver) Clear() error { + err := d.DB.Update(func(tx *bolt.Tx) error { + err := tx.DeleteBucket(d.bucket) + if err != nil { + d.log.Error("boltdb delete bucket", "error", err) + return err + } + + _, err = tx.CreateBucket(d.bucket) + if err != nil { + d.log.Error("boltdb create bucket", "error", err) + return err + } + + return nil + }) + + if err != nil { + d.log.Error("clear transaction failed", "error", err) + return err + } + + d.clearMu.Lock() + d.gc = sync.Map{} + d.clearMu.Unlock() + + return nil +} + // ========================= PRIVATE ================================= func (d *Driver) startGCLoop() { //nolint:gocognit @@ -382,6 +412,8 @@ func (d *Driver) startGCLoop() { //nolint:gocognit for { select { case <-t.C: + d.clearMu.RLock() + // calculate current time before loop started to be fair now := time.Now() d.gc.Range(func(key, value interface{}) bool { @@ -414,6 +446,8 @@ func (d *Driver) startGCLoop() { //nolint:gocognit } return true }) + + d.clearMu.RUnlock() case <-d.stop: err := d.DB.Close() if err != nil { diff --git a/plugins/kv/drivers/memcached/driver.go b/plugins/kv/drivers/memcached/driver.go index a2787d729..14e7c078c 100644 --- a/plugins/kv/drivers/memcached/driver.go +++ b/plugins/kv/drivers/memcached/driver.go @@ -237,3 +237,13 @@ func (d *Driver) Delete(keys ...string) error { } return nil } + +func (d *Driver) Clear() error { + err := d.client.DeleteAll() + if err != nil { + d.log.Error("flush_all operation failed", "error", err) + return err + } + + return nil +} diff --git a/plugins/kv/interface.go b/plugins/kv/interface.go index ffdbbe62c..5736a6a7a 100644 --- a/plugins/kv/interface.go +++ b/plugins/kv/interface.go @@ -22,9 +22,12 @@ type Storage interface { MExpire(items ...*kvv1.Item) error // TTL return the rest time to live for provided keys - // Not supported for the memcached and boltdb + // Not supported for the memcached TTL(keys ...string) (map[string]string, error) + // Clear clean the entire storage + Clear() error + // Delete one or multiple keys. Delete(keys ...string) error } diff --git a/plugins/kv/rpc.go b/plugins/kv/rpc.go index af763600d..3f7ba97cc 100644 --- a/plugins/kv/rpc.go +++ b/plugins/kv/rpc.go @@ -161,3 +161,19 @@ func (r *rpc) Delete(in *kvv1.Request, _ *kvv1.Response) error { return errors.E(op, errors.Errorf("no such storage: %s", in.GetStorage())) } + +// Clear clean the storage +func (r *rpc) Clear(in *kvv1.Request, _ *kvv1.Response) error { + const op = errors.Op("rcp_delete") + + if st, exists := r.storages[in.GetStorage()]; exists { + err := st.Clear() + if err != nil { + return errors.E(op, err) + } + + return nil + } + + return errors.E(op, errors.Errorf("no such storage: %s", in.GetStorage())) +} diff --git a/plugins/memory/kv.go b/plugins/memory/kv.go index 1cf031d13..c13c2314a 100644 --- a/plugins/memory/kv.go +++ b/plugins/memory/kv.go @@ -13,7 +13,8 @@ import ( ) type Driver struct { - heap sync.Map + clearMu sync.RWMutex + heap sync.Map // stop is used to stop keys GC and close boltdb connection stop chan struct{} log logger.Logger @@ -203,6 +204,14 @@ func (s *Driver) Delete(keys ...string) error { return nil } +func (s *Driver) Clear() error { + s.clearMu.Lock() + s.heap = sync.Map{} + s.clearMu.Unlock() + + return nil +} + // ================================== PRIVATE ====================================== func (s *Driver) gc() { @@ -213,6 +222,9 @@ func (s *Driver) gc() { ticker.Stop() return case now := <-ticker.C: + // mutes needed to clear the map + s.clearMu.RLock() + // check every second s.heap.Range(func(key, value interface{}) bool { v := value.(*kvv1.Item) @@ -231,6 +243,8 @@ func (s *Driver) gc() { } return true }) + + s.clearMu.RUnlock() } } } diff --git a/plugins/redis/kv.go b/plugins/redis/kv.go index 320b74437..2e4b9bfd4 100644 --- a/plugins/redis/kv.go +++ b/plugins/redis/kv.go @@ -240,3 +240,12 @@ func (d *Driver) TTL(keys ...string) (map[string]string, error) { } return m, nil } + +func (d *Driver) Clear() error { + fdb := d.universalClient.FlushDB(context.Background()) + if fdb.Err() != nil { + return fdb.Err() + } + + return nil +} diff --git a/tests/plugins/broadcast/broadcast_plugin_test.go b/tests/plugins/broadcast/broadcast_plugin_test.go index 2cd4b451c..0ec813f3a 100644 --- a/tests/plugins/broadcast/broadcast_plugin_test.go +++ b/tests/plugins/broadcast/broadcast_plugin_test.go @@ -205,7 +205,8 @@ func TestBroadcastSameSubscriber(t *testing.T) { cfg, &broadcast.Plugin{}, &rpcPlugin.Plugin{}, - mockLogger, + &logger.ZapLogger{}, + // mockLogger, &server.Plugin{}, &redis.Plugin{}, &websockets.Plugin{}, @@ -314,7 +315,8 @@ func TestBroadcastSameSubscriberGlobal(t *testing.T) { cfg, &broadcast.Plugin{}, &rpcPlugin.Plugin{}, - mockLogger, + &logger.ZapLogger{}, + // mockLogger, &server.Plugin{}, &redis.Plugin{}, &websockets.Plugin{}, diff --git a/tests/plugins/kv/storage_plugin_test.go b/tests/plugins/kv/storage_plugin_test.go index 1e466e066..ced1c5fe9 100644 --- a/tests/plugins/kv/storage_plugin_test.go +++ b/tests/plugins/kv/storage_plugin_test.go @@ -575,6 +575,53 @@ func testRPCMethods(t *testing.T) { err = client.Call("kv.Has", keysDel, ret) assert.NoError(t, err) assert.Len(t, ret.GetItems(), 0) + + dataClear := &payload.Request{ + Storage: "boltdb-rr", + Items: []*payload.Item{ + { + Key: "a", + Value: []byte("aa"), + }, + { + Key: "b", + Value: []byte("bb"), + }, + { + Key: "c", + Value: []byte("cc"), + }, + { + Key: "d", + Value: []byte("dd"), + }, + { + Key: "e", + Value: []byte("ee"), + }, + }, + } + + clear := &payload.Request{Storage: "boltdb-rr"} + + ret = &payload.Response{} + // Register 3 keys with values + err = client.Call("kv.Set", dataClear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 5) // should be 5 + + ret = &payload.Response{} + err = client.Call("kv.Clear", clear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 0) // should be 5 } func TestMemcached(t *testing.T) { @@ -790,6 +837,54 @@ func testRPCMethodsMemcached(t *testing.T) { err = client.Call("kv.Has", keysDel, ret) assert.NoError(t, err) assert.Len(t, ret.GetItems(), 0) + + dataClear := &payload.Request{ + Storage: "memcached-rr", + Items: []*payload.Item{ + { + Key: "a", + Value: []byte("aa"), + }, + { + Key: "b", + Value: []byte("bb"), + }, + { + Key: "c", + Value: []byte("cc"), + }, + { + Key: "d", + Value: []byte("dd"), + }, + { + Key: "e", + Value: []byte("ee"), + }, + }, + } + + clear := &payload.Request{Storage: "memcached-rr"} + + ret = &payload.Response{} + // Register 3 keys with values + err = client.Call("kv.Set", dataClear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 5) // should be 5 + + ret = &payload.Response{} + err = client.Call("kv.Clear", clear, ret) + assert.NoError(t, err) + + time.Sleep(time.Second * 2) + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 0) // should be 5 } func TestInMemory(t *testing.T) { @@ -1004,6 +1099,53 @@ func testRPCMethodsInMemory(t *testing.T) { err = client.Call("kv.Has", keysDel, ret) assert.NoError(t, err) assert.Len(t, ret.GetItems(), 0) + + dataClear := &payload.Request{ + Storage: "memory-rr", + Items: []*payload.Item{ + { + Key: "a", + Value: []byte("aa"), + }, + { + Key: "b", + Value: []byte("bb"), + }, + { + Key: "c", + Value: []byte("cc"), + }, + { + Key: "d", + Value: []byte("dd"), + }, + { + Key: "e", + Value: []byte("ee"), + }, + }, + } + + clear := &payload.Request{Storage: "memory-rr"} + + ret = &payload.Response{} + // Register 3 keys with values + err = client.Call("kv.Set", dataClear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 5) // should be 5 + + ret = &payload.Response{} + err = client.Call("kv.Clear", clear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 0) // should be 5 } func TestRedis(t *testing.T) { @@ -1354,4 +1496,51 @@ func testRPCMethodsRedis(t *testing.T) { err = client.Call("kv.Has", keysDel, ret) assert.NoError(t, err) assert.Len(t, ret.GetItems(), 0) + + dataClear := &payload.Request{ + Storage: "redis-rr", + Items: []*payload.Item{ + { + Key: "a", + Value: []byte("aa"), + }, + { + Key: "b", + Value: []byte("bb"), + }, + { + Key: "c", + Value: []byte("cc"), + }, + { + Key: "d", + Value: []byte("dd"), + }, + { + Key: "e", + Value: []byte("ee"), + }, + }, + } + + clear := &payload.Request{Storage: "redis-rr"} + + ret = &payload.Response{} + // Register 3 keys with values + err = client.Call("kv.Set", dataClear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 5) // should be 5 + + ret = &payload.Response{} + err = client.Call("kv.Clear", clear, ret) + assert.NoError(t, err) + + ret = &payload.Response{} + err = client.Call("kv.Has", dataClear, ret) + assert.NoError(t, err) + assert.Len(t, ret.GetItems(), 0) // should be 5 } diff --git a/tests/plugins/websockets/websocket_plugin_test.go b/tests/plugins/websockets/websocket_plugin_test.go index 5ed0c3f34..53b6a572a 100644 --- a/tests/plugins/websockets/websocket_plugin_test.go +++ b/tests/plugins/websockets/websocket_plugin_test.go @@ -645,7 +645,7 @@ func RPCWsPubAsync(port string) func(t *testing.T) { return func(t *testing.T) { da := websocket.Dialer{ Proxy: http.ProxyFromEnvironment, - HandshakeTimeout: time.Second * 18, + HandshakeTimeout: time.Second * 20, } connURL := url.URL{Scheme: "ws", Host: "localhost:" + port, Path: "/ws"} @@ -654,9 +654,32 @@ func RPCWsPubAsync(port string) func(t *testing.T) { assert.NoError(t, err) defer func() { - _ = resp.Body.Close() + if resp != nil && resp.Body != nil { + _ = resp.Body.Close() + } + }() + + go func() { + messagesToVerify := make([]string, 0, 10) + messagesToVerify = append(messagesToVerify, `{"topic":"@join","payload":["foo","foo2"]}`) + messagesToVerify = append(messagesToVerify, `{"topic":"foo","payload":"hello, PHP"}`) + messagesToVerify = append(messagesToVerify, `{"topic":"@leave","payload":["foo"]}`) + messagesToVerify = append(messagesToVerify, `{"topic":"foo2","payload":"hello, PHP2"}`) + i := 0 + for { + _, msg, err2 := c.ReadMessage() + retMsg := utils.AsString(msg) + assert.NoError(t, err2) + assert.Equal(t, messagesToVerify[i], retMsg) + i++ + if i == 3 { + return + } + } }() + time.Sleep(time.Second) + d, err := json.Marshal(messageWS("join", []byte("hello websockets"), "foo", "foo2")) if err != nil { panic(err) @@ -665,20 +688,11 @@ func RPCWsPubAsync(port string) func(t *testing.T) { err = c.WriteMessage(websocket.BinaryMessage, d) assert.NoError(t, err) - _, msg, err := c.ReadMessage() - retMsg := utils.AsString(msg) - assert.NoError(t, err) + time.Sleep(time.Second) - // subscription done - assert.Equal(t, `{"topic":"@join","payload":["foo","foo2"]}`, retMsg) + publishAsync(t, "foo") - publishAsync(t, "placeholder", "foo") - - // VERIFY a makeMessage - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - assert.Equal(t, "{\"topic\":\"foo\",\"payload\":\"hello, PHP\"}", retMsg) + time.Sleep(time.Second) // //// LEAVE foo ///////// d, err = json.Marshal(messageWS("leave", []byte("hello websockets"), "foo")) @@ -689,27 +703,16 @@ func RPCWsPubAsync(port string) func(t *testing.T) { err = c.WriteMessage(websocket.BinaryMessage, d) assert.NoError(t, err) - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - - // subscription done - assert.Equal(t, `{"topic":"@leave","payload":["foo"]}`, retMsg) + time.Sleep(time.Second) // TRY TO PUBLISH TO UNSUBSCRIBED TOPIC - publishAsync(t, "placeholder", "foo") + publishAsync(t, "foo") go func() { - time.Sleep(time.Second * 3) - publishAsync(t, "placeholder", "foo2") + time.Sleep(time.Second * 5) + publishAsync(t, "foo2") }() - // should be only makeMessage from the subscribed foo0 topic - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - assert.Equal(t, "{\"topic\":\"foo2\",\"payload\":\"hello, PHP\"}", retMsg) - err = c.WriteControl(websocket.CloseMessage, nil, time.Time{}) assert.NoError(t, err) } @@ -733,6 +736,27 @@ func RPCWsPub(port string) func(t *testing.T) { } }() + go func() { + messagesToVerify := make([]string, 0, 10) + messagesToVerify = append(messagesToVerify, `{"topic":"@join","payload":["foo","foo2"]}`) + messagesToVerify = append(messagesToVerify, `{"topic":"foo","payload":"hello, PHP"}`) + messagesToVerify = append(messagesToVerify, `{"topic":"@leave","payload":["foo"]}`) + messagesToVerify = append(messagesToVerify, `{"topic":"foo2","payload":"hello, PHP2"}`) + i := 0 + for { + _, msg, err2 := c.ReadMessage() + retMsg := utils.AsString(msg) + assert.NoError(t, err2) + assert.Equal(t, messagesToVerify[i], retMsg) + i++ + if i == 3 { + return + } + } + }() + + time.Sleep(time.Second) + d, err := json.Marshal(messageWS("join", []byte("hello websockets"), "foo", "foo2")) if err != nil { panic(err) @@ -741,20 +765,11 @@ func RPCWsPub(port string) func(t *testing.T) { err = c.WriteMessage(websocket.BinaryMessage, d) assert.NoError(t, err) - _, msg, err := c.ReadMessage() - retMsg := utils.AsString(msg) - assert.NoError(t, err) - - // subscription done - assert.Equal(t, `{"topic":"@join","payload":["foo","foo2"]}`, retMsg) + time.Sleep(time.Second) publish("", "foo") - // VERIFY a makeMessage - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - assert.Equal(t, "{\"topic\":\"foo\",\"payload\":\"hello, PHP\"}", retMsg) + time.Sleep(time.Second) // //// LEAVE foo ///////// d, err = json.Marshal(messageWS("leave", []byte("hello websockets"), "foo")) @@ -765,12 +780,7 @@ func RPCWsPub(port string) func(t *testing.T) { err = c.WriteMessage(websocket.BinaryMessage, d) assert.NoError(t, err) - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - - // subscription done - assert.Equal(t, `{"topic":"@leave","payload":["foo"]}`, retMsg) + time.Sleep(time.Second) // TRY TO PUBLISH TO UNSUBSCRIBED TOPIC publish("", "foo") @@ -780,12 +790,6 @@ func RPCWsPub(port string) func(t *testing.T) { publish2(t, "", "foo2") }() - // should be only makeMessage from the subscribed foo2 topic - _, msg, err = c.ReadMessage() - retMsg = utils.AsString(msg) - assert.NoError(t, err) - assert.Equal(t, "{\"topic\":\"foo2\",\"payload\":\"hello, PHP2\"}", retMsg) - err = c.WriteControl(websocket.CloseMessage, nil, time.Time{}) assert.NoError(t, err) } @@ -849,7 +853,7 @@ func RPCWsDeny(port string) func(t *testing.T) { // --------------------------------------------------------------------------------------------------- -func publish(command string, topics ...string) { +func publish(topics ...string) { conn, err := net.Dial("tcp", "127.0.0.1:6001") if err != nil { panic(err) @@ -858,13 +862,13 @@ func publish(command string, topics ...string) { client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) ret := &websocketsv1.Response{} - err = client.Call("broadcast.Publish", makeMessage(command, []byte("hello, PHP"), topics...), ret) + err = client.Call("broadcast.Publish", makeMessage([]byte("hello, PHP"), topics...), ret) if err != nil { panic(err) } } -func publishAsync(t *testing.T, command string, topics ...string) { +func publishAsync(t *testing.T, topics ...string) { conn, err := net.Dial("tcp", "127.0.0.1:6001") if err != nil { panic(err) @@ -873,12 +877,12 @@ func publishAsync(t *testing.T, command string, topics ...string) { client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) ret := &websocketsv1.Response{} - err = client.Call("broadcast.PublishAsync", makeMessage(command, []byte("hello, PHP"), topics...), ret) + err = client.Call("broadcast.PublishAsync", makeMessage([]byte("hello, PHP"), topics...), ret) assert.NoError(t, err) assert.True(t, ret.Ok) } -func publish2(t *testing.T, command string, topics ...string) { +func publish2(t *testing.T, topics ...string) { conn, err := net.Dial("tcp", "127.0.0.1:6001") if err != nil { panic(err) @@ -887,7 +891,7 @@ func publish2(t *testing.T, command string, topics ...string) { client := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) ret := &websocketsv1.Response{} - err = client.Call("broadcast.Publish", makeMessage(command, []byte("hello, PHP2"), topics...), ret) + err = client.Call("broadcast.Publish", makeMessage([]byte("hello, PHP2"), topics...), ret) assert.NoError(t, err) assert.True(t, ret.Ok) } @@ -900,12 +904,11 @@ func messageWS(command string, payload []byte, topics ...string) *websocketsv1.M } } -func makeMessage(command string, payload []byte, topics ...string) *websocketsv1.Request { +func makeMessage(payload []byte, topics ...string) *websocketsv1.Request { m := &websocketsv1.Request{ Messages: []*websocketsv1.Message{ { Topics: topics, - Command: command, Payload: payload, }, },