- Add new internal plugin - channel. Which used to deliver messages from

  the ws plugin to the http directly

Signed-off-by: Valery Piashchynski <[email protected]>

9 files changed, 72 insertions, 8 deletions

diff --git a/tests/plugins/config/config_test.go b/tests/plugins/config/config_test.go
index b6063cec..3cf026bd 100755
--- a/tests/plugins/config/config_test.go
+++ b/tests/plugins/config/config_test.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	endure "github.com/spiral/endure/pkg/container"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
 	"github.com/spiral/roadrunner/v2/plugins/rpc"
@@ -33,6 +34,11 @@ func TestViperProvider_Init(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	err = container.Register(&channel.Plugin{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	err = container.Init()
 	if err != nil {
 		t.Fatal(err)
@@ -82,6 +88,7 @@ func TestConfigOverwriteFail(t *testing.T) {
 		&rpc.Plugin{},
 		vp,
 		&Foo2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -103,6 +110,7 @@ func TestConfigOverwriteValid(t *testing.T) {
 		&logger.ZapLogger{},
 		&rpc.Plugin{},
 		vp,
+		&channel.Plugin{},
 		&Foo2{},
 	)
 	assert.NoError(t, err)
@@ -155,6 +163,7 @@ func TestConfigEnvVariables(t *testing.T) {
 		&rpc.Plugin{},
 		vp,
 		&Foo2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -206,6 +215,7 @@ func TestConfigEnvVariablesFail(t *testing.T) {
 		&rpc.Plugin{},
 		vp,
 		&Foo2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -237,6 +247,11 @@ func TestConfigProvider_GeneralSection(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	err = container.Register(&channel.Plugin{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	err = container.Init()
 	if err != nil {
 		t.Fatal(err)
diff --git a/tests/plugins/gzip/plugin_test.go b/tests/plugins/gzip/plugin_test.go
index 844fd411..5294e672 100644
--- a/tests/plugins/gzip/plugin_test.go
+++ b/tests/plugins/gzip/plugin_test.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/gzip"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
@@ -35,6 +36,7 @@ func TestGzipPlugin(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -128,6 +130,7 @@ func TestMiddlewareNotExist(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/headers/headers_plugin_test.go b/tests/plugins/headers/headers_plugin_test.go
index 49d86b00..e4903335 100644
--- a/tests/plugins/headers/headers_plugin_test.go
+++ b/tests/plugins/headers/headers_plugin_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	endure "github.com/spiral/endure/pkg/container"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/headers"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
@@ -34,6 +35,7 @@ func TestHeadersInit(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&headers.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -100,6 +102,7 @@ func TestRequestHeaders(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&headers.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -185,6 +188,7 @@ func TestResponseHeaders(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&headers.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -271,6 +275,7 @@ func TestCORSHeaders(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&headers.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/http/http_plugin_test.go b/tests/plugins/http/http_plugin_test.go
index 128eec26..aa57077d 100644
--- a/tests/plugins/http/http_plugin_test.go
+++ b/tests/plugins/http/http_plugin_test.go
@@ -24,6 +24,7 @@ import (
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
 	"github.com/spiral/roadrunner/v2/pkg/events"
 	"github.com/spiral/roadrunner/v2/pkg/process"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/gzip"
 	"github.com/spiral/roadrunner/v2/plugins/informer"
@@ -62,6 +63,7 @@ func TestHTTPInit(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -126,6 +128,7 @@ func TestHTTPNoConfigSection(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -193,6 +196,7 @@ func TestHTTPInformerReset(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&informer.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -315,6 +319,7 @@ func TestSSL(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -451,6 +456,7 @@ func TestSSLRedirect(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -540,6 +546,7 @@ func TestSSLPushPipes(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -630,6 +637,7 @@ func TestFastCGI_RequestUri(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -724,6 +732,7 @@ func TestH2CUpgrade(t *testing.T) {
 		mockLogger,
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -815,6 +824,7 @@ func TestH2C(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -907,6 +917,7 @@ func TestHttpMiddleware(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&PluginMiddleware{},
 		&PluginMiddleware2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1053,6 +1064,7 @@ logs:
 		&httpPlugin.Plugin{},
 		&PluginMiddleware{},
 		&PluginMiddleware2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1138,6 +1150,7 @@ func TestHttpEnvVariables(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&PluginMiddleware{},
 		&PluginMiddleware2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1225,6 +1238,7 @@ func TestHttpBrokenPipes(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&PluginMiddleware{},
 		&PluginMiddleware2{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1286,6 +1300,7 @@ func TestHTTPSupervisedPool(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&informer.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1488,6 +1503,7 @@ func TestHTTPBigRequestSize(t *testing.T) {
 		&logger.ZapLogger{},
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1580,6 +1596,7 @@ func TestStaticEtagPlugin(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1678,6 +1695,7 @@ func TestStaticPluginSecurity(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1827,6 +1845,7 @@ func TestStaticPlugin(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -1941,6 +1960,7 @@ func TestStaticDisabled_Error(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 	assert.Error(t, cont.Init())
@@ -1962,6 +1982,7 @@ func TestStaticFilesDisabled(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -2054,6 +2075,7 @@ func TestStaticFilesForbid(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&gzip.Plugin{},
 		&static.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/logger/logger_test.go b/tests/plugins/logger/logger_test.go
index d2877781..f63a6a5d 100644
--- a/tests/plugins/logger/logger_test.go
+++ b/tests/plugins/logger/logger_test.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/http"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
@@ -98,6 +99,7 @@ func TestLoggerRawErr(t *testing.T) {
 		mockLogger,
 		&server.Plugin{},
 		&http.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -224,6 +226,7 @@ func TestLoggerNoConfig2(t *testing.T) {
 		&logger.ZapLogger{},
 		&http.Plugin{},
 		&server.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/metrics/metrics_test.go b/tests/plugins/metrics/metrics_test.go
index 8be567ec..48c01f24 100644
--- a/tests/plugins/metrics/metrics_test.go
+++ b/tests/plugins/metrics/metrics_test.go
@@ -15,6 +15,7 @@ import (
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
@@ -144,6 +145,7 @@ func TestMetricsIssue571(t *testing.T) {
 		&server.Plugin{},
 		mockLogger,
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/reload/reload_plugin_test.go b/tests/plugins/reload/reload_plugin_test.go
index 6db7b6d0..41c9c92f 100644
--- a/tests/plugins/reload/reload_plugin_test.go
+++ b/tests/plugins/reload/reload_plugin_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/golang/mock/gomock"
 	endure "github.com/spiral/endure/pkg/container"
 	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
 	"github.com/spiral/roadrunner/v2/plugins/reload"
@@ -65,6 +66,7 @@ func TestReloadInit(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&reload.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -161,6 +163,7 @@ func TestReloadHugeNumberOfFiles(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&reload.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -270,6 +273,7 @@ func TestReloadFilterFileExt(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&reload.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -400,6 +404,7 @@ func TestReloadCopy100(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&reload.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -677,6 +682,7 @@ func TestReloadNoRecursion(t *testing.T) {
 		&httpPlugin.Plugin{},
 		&reload.Plugin{},
 		&resetter.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/status/plugin_test.go b/tests/plugins/status/plugin_test.go
index 663f4ee3..06983199 100644
--- a/tests/plugins/status/plugin_test.go
+++ b/tests/plugins/status/plugin_test.go
@@ -14,6 +14,7 @@ import (
 
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
@@ -38,6 +39,7 @@ func TestStatusHttp(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&status.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -125,6 +127,7 @@ func TestStatusRPC(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&status.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -204,6 +207,7 @@ func TestReadyHttp(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&status.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -291,6 +295,7 @@ func TestReadinessRPCWorkerNotReady(t *testing.T) {
 		&server.Plugin{},
 		&httpPlugin.Plugin{},
 		&status.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
diff --git a/tests/plugins/websockets/websocket_plugin_test.go b/tests/plugins/websockets/websocket_plugin_test.go
index 087d1bc9..61ef186b 100644
--- a/tests/plugins/websockets/websocket_plugin_test.go
+++ b/tests/plugins/websockets/websocket_plugin_test.go
@@ -16,6 +16,7 @@ import (
 	json "github.com/json-iterator/go"
 	endure "github.com/spiral/endure/pkg/container"
 	goridgeRpc "github.com/spiral/goridge/v3/pkg/rpc"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	httpPlugin "github.com/spiral/roadrunner/v2/plugins/http"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
@@ -30,16 +31,16 @@ import (
 
 type Msg struct {
 	// Topic message been pushed into.
-	Topics_ []string `json:"topic"`
+	Topics []string `json:"topic"`
 
 	// Command (join, leave, headers)
-	Command_ string `json:"command"`
+	Command string `json:"command"`
 
 	// Broker (redis, memory)
-	Broker_ string `json:"broker"`
+	Broker string `json:"broker"`
 
 	// Payload to be broadcasted
-	Payload_ []byte `json:"payload"`
+	Payload []byte `json:"payload"`
 }
 
 func TestBroadcastInit(t *testing.T) {
@@ -59,6 +60,7 @@ func TestBroadcastInit(t *testing.T) {
 		&redis.Plugin{},
 		&websockets.Plugin{},
 		&httpPlugin.Plugin{},
+		&channel.Plugin{},
 	)
 
 	assert.NoError(t, err)
@@ -168,6 +170,7 @@ func TestWSRedisAndMemory(t *testing.T) {
 		&websockets.Plugin{},
 		&httpPlugin.Plugin{},
 		&memory.Plugin{},
+		&channel.Plugin{},
 	)
 	assert.NoError(t, err)
 
@@ -504,9 +507,9 @@ func publish2(command string, broker string, topics ...string) {
 
 func message(command string, broker string, payload []byte, topics ...string) *Msg {
 	return &Msg{
-		Topics_:  topics,
-		Command_: command,
-		Broker_:  broker,
-		Payload_: payload,
+		Topics:  topics,
+		Command: command,
+		Broker:  broker,
+		Payload: payload,
 	}
 }