6 files changed, 184 insertions, 0 deletions
diff --git a/plugins/broadcast/root/tests/.rr.yaml b/plugins/broadcast/root/tests/.rr.yaml
new file mode 100644
index 00000000..c35a12fc
--- /dev/null
+++ b/plugins/broadcast/root/tests/.rr.yaml
@@ -0,0 +1,2 @@
+broadcast:
+  redis.addr: "localhost:6379"
+\ No newline at end of file
diff --git a/plugins/broadcast/root/tests/Broadcast/BroadcastTest.php b/plugins/broadcast/root/tests/Broadcast/BroadcastTest.php
new file mode 100644
index 00000000..d6014bf0
--- /dev/null
+++ b/plugins/broadcast/root/tests/Broadcast/BroadcastTest.php
@@ -0,0 +1,56 @@
+<?php
+
+/**
+ * Spiral Framework.
+ *
+ * @license   MIT
+ * @author    Anton Titov (Wolfy-J)
+ */
+
+declare(strict_types=1);
+
+namespace Spiral\Broadcast\Tests;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\Broadcast\Broadcast;
+use Spiral\Broadcast\Exception\BroadcastException;
+use Spiral\Broadcast\Message;
+use Spiral\Goridge\RPC;
+use Spiral\Goridge\SocketRelay;
+
+class BroadcastTest extends TestCase
+{
+    public function testBroadcast(): void
+    {
+        $rpc = new RPC(new SocketRelay('localhost', 6001));
+        $br = new Broadcast($rpc);
+
+        $br->publish(
+            new Message('tests/topic', 'hello'),
+            new Message('tests/123', ['key' => 'value'])
+        );
+
+        while (filesize(__DIR__ . '/../log.txt') < 40) {
+            clearstatcache(true, __DIR__ . '/../log.txt');
+            usleep(1000);
+        }
+
+        clearstatcache(true, __DIR__ . '/../log.txt');
+        $content = file_get_contents(__DIR__ . '/../log.txt');
+
+        $this->assertSame('tests/topic: "hello"
+tests/123: {"key":"value"}
+', $content);
+    }
+
+    public function testBroadcastException(): void
+    {
+        $rpc = new RPC(new SocketRelay('localhost', 6002));
+        $br = new Broadcast($rpc);
+
+        $this->expectException(BroadcastException::class);
+        $br->publish(
+            new Message('topic', 'hello')
+        );
+    }
+}
diff --git a/plugins/broadcast/root/tests/Broadcast/MessageTest.php b/plugins/broadcast/root/tests/Broadcast/MessageTest.php
new file mode 100644
index 00000000..dd9e1cc3
--- /dev/null
+++ b/plugins/broadcast/root/tests/Broadcast/MessageTest.php
@@ -0,0 +1,24 @@
+<?php
+
+/**
+ * Spiral Framework.
+ *
+ * @license   MIT
+ * @author    Anton Titov (Wolfy-J)
+ */
+
+declare(strict_types=1);
+
+namespace Spiral\Broadcast\Tests;
+
+use PHPUnit\Framework\TestCase;
+use Spiral\Broadcast\Message;
+
+class MessageTest extends TestCase
+{
+    public function testSerialize(): void
+    {
+        $m = new Message('topic', ['hello' => 'world']);
+        $this->assertSame('{"topic":"topic","payload":{"hello":"world"}}', json_encode($m));
+    }
+}
diff --git a/plugins/broadcast/root/tests/bootstrap.php b/plugins/broadcast/root/tests/bootstrap.php
new file mode 100644
index 00000000..d0dfb88b
--- /dev/null
+++ b/plugins/broadcast/root/tests/bootstrap.php
@@ -0,0 +1,15 @@
+<?php
+
+/**
+ * Spiral Framework, SpiralScout LLC.
+ *
+ * @author    Anton Titov (Wolfy-J)
+ */
+
+declare(strict_types=1);
+
+error_reporting(E_ALL | E_STRICT);
+ini_set('display_errors', 'stderr');
+
+//Composer
+require dirname(__DIR__) . '/vendor_php/autoload.php';
diff --git a/plugins/broadcast/root/tests/docker-compose.yml b/plugins/broadcast/root/tests/docker-compose.yml
new file mode 100644
index 00000000..123aa9b9
--- /dev/null
+++ b/plugins/broadcast/root/tests/docker-compose.yml
@@ -0,0 +1,9 @@
+version: '3'
+
+services:
+  redis:
+    image: 'bitnami/redis:latest'
+    environment:
+      - ALLOW_EMPTY_PASSWORD=yes
+    ports:
+      - "6379:6379"
+\ No newline at end of file
diff --git a/plugins/broadcast/root/tests/go-client.go b/plugins/broadcast/root/tests/go-client.go
new file mode 100644
index 00000000..21442a01
--- /dev/null
+++ b/plugins/broadcast/root/tests/go-client.go
@@ -0,0 +1,78 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/spiral/broadcast/v2"
+	rr "github.com/spiral/roadrunner/cmd/rr/cmd"
+	"github.com/spiral/roadrunner/service/rpc"
+	"golang.org/x/sync/errgroup"
+)
+
+type logService struct {
+	broadcast *broadcast.Service
+	stop      chan interface{}
+}
+
+func (l *logService) Init(service *broadcast.Service) (bool, error) {
+	l.broadcast = service
+
+	return true, nil
+}
+
+func (l *logService) Serve() error {
+	l.stop = make(chan interface{})
+
+	client := l.broadcast.NewClient()
+	if err := client.SubscribePattern("tests/*"); err != nil {
+		return err
+	}
+
+	logFile, _ := os.Create("log.txt")
+
+	g := &errgroup.Group{}
+	g.Go(func() error {
+		for msg := range client.Channel() {
+			_, err := logFile.Write([]byte(fmt.Sprintf(
+				"%s: %s\n",
+				msg.Topic,
+				string(msg.Payload),
+			)))
+			if err != nil {
+				return err
+			}
+
+			err = logFile.Sync()
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+
+	<-l.stop
+	err := logFile.Close()
+	if err != nil {
+		return err
+	}
+
+	err = client.Close()
+	if err != nil {
+		return err
+	}
+
+	return g.Wait()
+}
+
+func (l *logService) Stop() {
+	close(l.stop)
+}
+
+func main() {
+	rr.Container.Register(rpc.ID, &rpc.Service{})
+	rr.Container.Register(broadcast.ID, &broadcast.Service{})
+	rr.Container.Register("log", &logService{})
+
+	rr.Execute()
+}