From fcda08498e8f914bbd0798da898818cd5d0e4348 Mon Sep 17 00:00:00 2001
From: Valery Piashchynski <piashchynski.valery@gmail.com>
Date: Sat, 29 May 2021 00:24:30 +0300
Subject: - Add new internal plugin - channel. Which used to deliver messages
 from   the ws plugin to the http directly

Signed-off-by: Valery Piashchynski <piashchynski.valery@gmail.com>
---
 plugins/http/channel.go | 28 ++++++++++++++++++++++++++++
 plugins/http/plugin.go  |  9 ++++++++-
 2 files changed, 36 insertions(+), 1 deletion(-)
 create mode 100644 plugins/http/channel.go

(limited to 'plugins/http')

diff --git a/plugins/http/channel.go b/plugins/http/channel.go
new file mode 100644
index 00000000..42b73730
--- /dev/null
+++ b/plugins/http/channel.go
@@ -0,0 +1,28 @@
+package http
+
+import (
+	"net/http"
+)
+
+// messages method used to read messages from the ws plugin with the auth requests for the topics and server
+func (p *Plugin) messages() {
+	for msg := range p.hub.ReceiveCh() {
+		p.RLock()
+		// msg here is the structure with http.ResponseWriter and http.Request
+		rmsg := msg.(struct {
+			RW  http.ResponseWriter
+			Req *http.Request
+		})
+
+		p.handler.ServeHTTP(rmsg.RW, rmsg.Req)
+
+		p.hub.SendCh() <- struct {
+			RW  http.ResponseWriter
+			Req *http.Request
+		}{
+			rmsg.RW,
+			rmsg.Req,
+		}
+		p.RUnlock()
+	}
+}
diff --git a/plugins/http/plugin.go b/plugins/http/plugin.go
index 8bcffb63..38b3621f 100644
--- a/plugins/http/plugin.go
+++ b/plugins/http/plugin.go
@@ -14,6 +14,7 @@ import (
 	"github.com/spiral/roadrunner/v2/pkg/process"
 	"github.com/spiral/roadrunner/v2/pkg/worker"
 	handler "github.com/spiral/roadrunner/v2/pkg/worker_handler"
+	"github.com/spiral/roadrunner/v2/plugins/channel"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/http/attributes"
 	httpConfig "github.com/spiral/roadrunner/v2/plugins/http/config"
@@ -67,11 +68,14 @@ type Plugin struct {
 	http  *http.Server
 	https *http.Server
 	fcgi  *http.Server
+
+	// message bus
+	hub channel.Hub
 }
 
 // Init must return configure svc and return true if svc hasStatus enabled. Must return error in case of
 // misconfiguration. Services must not be used without proper configuration pushed first.
-func (p *Plugin) Init(cfg config.Configurer, rrLogger logger.Logger, server server.Server) error {
+func (p *Plugin) Init(cfg config.Configurer, rrLogger logger.Logger, server server.Server, channel channel.Hub) error {
 	const op = errors.Op("http_plugin_init")
 	if !cfg.Has(PluginName) {
 		return errors.E(op, errors.Disabled)
@@ -105,6 +109,9 @@ func (p *Plugin) Init(cfg config.Configurer, rrLogger logger.Logger, server serv
 
 	p.cfg.Env[RrMode] = "http"
 	p.server = server
+	p.hub = channel
+
+	go p.messages()
 
 	return nil
 }
-- 
cgit v1.2.3