Added a way to override the send call with a hook.

Changes

@@ -1,6 +1,11 @@
 Revision History for ConsumerGroups
 
 {{$NEXT}}
+
+0.07      2024-05-21 12:30:18+00:00 UTC
+          Allow send method to be replaced with a hook
+
+0.06      2024-03-28 11:52:48+00:00 UTC
           Allow timeout to be extended vs published deadline
 
 0.05      2023-11-07 07:54:26+00:00 UTC

Makefile.PL

@@ -46,7 +46,7 @@ my %WriteMakefileArgs = (
     "Test::More" => "0.98",
     "Test::NoTabs" => 0
   },
-  "VERSION" => "0.06",
+  "VERSION" => "0.08",
   "test" => {
     "TESTS" => "t/*.t"
   }

README.md

@@ -4,7 +4,7 @@ Mojo::WebSocketProxy::Backend::ConsumerGroups - Class for communication with bac
 
 # VERSION
 
-version 0.05
+version 0.07
 
 # DESCRIPTION
 
@@ -102,19 +102,6 @@ By using redis subscription, we subscribe on channel for receiving responses fro
 We'll use uniq id generated by [whoami](https://metacpan.org/pod/whoami) as subscription channel.
 Subscription will be done only once within first request to backend server.
 
-# SEE ALSO
-
-[Mojolicious::Plugin::WebSocketProxy](https://metacpan.org/pod/Mojolicious%3A%3APlugin%3A%3AWebSocketProxy),
-[Mojo::WebSocketProxy](https://metacpan.org/pod/Mojo%3A%3AWebSocketProxy)
-[Mojo::WebSocketProxy::Backend](https://metacpan.org/pod/Mojo%3A%3AWebSocketProxy%3A%3ABackend),
-[Mojo::WebSocketProxy::Dispatcher](https://metacpan.org/pod/Mojo%3A%3AWebSocketProxy%3A%3ADispatcher),
-[Mojo::WebSocketProxy::Config](https://metacpan.org/pod/Mojo%3A%3AWebSocketProxy%3A%3AConfig)
-[Mojo::WebSocketProxy::Parser](https://metacpan.org/pod/Mojo%3A%3AWebSocketProxy%3A%3AParser)
-
-# COPYRIGHT AND LICENSE
-
-Copyright (C) 2022 deriv.com
-
 # AUTHOR
 
 DERIV <[email protected]>

lib/Mojo/WebSocketProxy/Backend/ConsumerGroups.pm

@@ -18,7 +18,7 @@ use parent qw(Mojo::WebSocketProxy::Backend);
 
 no indirect;
 
-our $VERSION = '0.06';
+our $VERSION = '0.08';
 
 __PACKAGE__->register_type('consumer_groups');
 
@@ -211,6 +211,7 @@ sub call_rpc {
     my $before_get_rpc_response_hooks = delete($req_storage->{before_get_rpc_response}) || [];
     my $after_got_rpc_response_hooks  = delete($req_storage->{after_got_rpc_response})  || [];
     my $before_call_hooks             = delete($req_storage->{before_call})             || [];
+    my $send_func                     = delete($req_storage->{send_func})               || (sub { $c->send(@_); });
     my $rpc_failure_cb                = delete($req_storage->{rpc_failure_cb});
     my $rpc_timeout_extend_offset     = delete($req_storage->{rpc_timeout_extend_offset});
     my $rpc_timeout_extend_percentage = delete($req_storage->{rpc_timeout_extend_percentage});
@@ -221,7 +222,7 @@ sub call_rpc {
     foreach my $hook ($before_call_hooks->@*) { $hook->($c, $req_storage) }
 
     my $category_deadline = $self->_rpc_category_timeout($req_storage->{category});
-    my $category_timeout = $self->_rpc_category_timeout($req_storage->{category});
+    my $category_timeout  = $self->_rpc_category_timeout($req_storage->{category});
 
     if ($rpc_timeout_extend_percentage) {
         $category_timeout += $category_timeout * ($rpc_timeout_extend_percentage / 100);
@@ -257,13 +258,13 @@ sub call_rpc {
 
                 return Future->done if $block_response;
                 $api_response = $c->wsp_error($msg_type, 'CallError', 'Sorry, an error occurred while processing your request.');
-                $c->send({json => $api_response}, $req_storage);
+                $send_func->({json => $api_response}, $req_storage);
                 return Future->done;
             }
 
             $api_response = $rpc_response_cb->($result->result);
             return Future->done if $block_response || !$api_response;
-            $c->send({json => $api_response}, $req_storage);
+            $send_func->({json => $api_response}, $req_storage);
 
             return Future->done;
         }
@@ -288,7 +289,7 @@ sub call_rpc {
 
             return Future->done if $block_response || !$api_response;
 
-            $c->send({json => $api_response}, $req_storage);
+            $send_func->({json => $api_response}, $req_storage);
         })->retain;
 
     return;