Configured message brocker

.gitlab-ci.yml

@@ -2,7 +2,7 @@ image: docker:20.10.9
 
 variables:
   DOCKER_TLS_CERTDIR: "/certs"
-  IMAGE_VER: 2.5.4
+  IMAGE_VER: 2.6.0
 
 services:
   - docker:20.10.9-dind

lib/CF/Controller/Websockets.pm

@@ -23,11 +23,20 @@ sub main {
         $c->res->headers->add('Sec-WebSocket-Extensions' => 'permessage-deflate');
     }
 
-    $c->redis->pubsub->listen(
-        notify => sub($pubsub, $payload) {
+    my $pubsub;
+
+    if ($c->cfg->{message_broker} eq 'redis') {
+        $pubsub = $c->redis->pubsub;
+        $pubsub->listen(notify => sub($pubsub, $payload) {
             $c->send(decode("UTF-8", $payload));
-        }
-    );
+        });
+    }
+    else {
+        $pubsub = Mojo::Pg::PubSub->new(pg => $c->pg);
+        $pubsub->listen(notify => sub($pubsub, $payload) {
+            $c->send($payload);
+        });
+    }
 
     $c->on(json => sub( $c, $message ) {
         if ( $message->{event} eq 'KEEPALIVE' ) {
@@ -99,7 +108,7 @@ sub main {
     });
 
     $c->on(finish => sub ($c, $code, $reason = undef) {
-        $c->redis->pubsub->unlisten('notify');
+        $pubsub->unlisten('notify');
         $c->app->log->debug("WebSocket closed with status $code");
     });
 }

lib/CF/Helpers/Core.pm

@@ -203,12 +203,19 @@ sub register ($class, $self, $conf) {
 
     $self->helper( notify => sub ( $c, $event, $payload ) {
 
-        my $content = encode_json ({
+        my $content = {
             event   => $event,
             payload => $payload,
-        });
+        };
+
+        if  ($self->cfg->{message_broker} eq 'redis' ) {
+            $self->redis->pubsub->notify( notify => encode_json($content) );
+        }
+        else {
+            my $pubsub = Mojo::Pg::PubSub->new(pg => $self->pg);
+            $pubsub->json('notify')->notify( notify => $content );
+        }
 
-        $self->redis->pubsub->notify( notify => $content );
     });
 }