Added compress options support for publishers

JustSnow · JustSnow · commit 932d2c66378d · 2026-03-25T13:04:57.000+03:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,10 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [6.11.0] - 2026-03-25
+### Added
+- Ability to specify option for compression when publish data
+
 ## [6.10.0] - 2025-11-21
 ### Added
 - Add on_first_sync callback
diff --git a/Gemfile b/Gemfile
@@ -19,3 +19,5 @@ gem "bundler"
 gem "ostruct"
 gem "pry"
 gem "rake"
+
+gem "rabbit_messaging", git: "git@github.com:umbrellio/rabbit_messaging.git", branch: "compression-for-publisher-and-reciver"
diff --git a/Gemfile.lock b/Gemfile.lock
@@ -1,9 +1,20 @@
+GIT
+  remote: git@github.com:umbrellio/rabbit_messaging.git
+  revision: 2a31c6878f012fd3a31f538a9d1ac2d0670a2e50
+  branch: compression-for-publisher-and-reciver
+  specs:
+    rabbit_messaging (1.8.0)
+      bunny (~> 2.0)
+      kicks
+      msgpack
+      zlib
+
 PATH
   remote: .
   specs:
-    table_sync (6.10.0)
+    table_sync (6.11.0)
       memery
-      rabbit_messaging (>= 1.7.0)
+      rabbit_messaging (>= 1.8.0)
       rails
       self_data
 
@@ -136,6 +147,7 @@ GEM
     mini_mime (1.1.5)
     mini_portile2 (2.8.9)
     minitest (5.25.5)
+    msgpack (1.8.0)
     net-imap (0.5.12)
       date
       net-protocol
@@ -181,9 +193,6 @@ GEM
     psych (5.2.6)
       date
       stringio
-    rabbit_messaging (1.7.0)
-      bunny (~> 2.0)
-      kicks
     racc (1.8.1)
     rack (3.1.15)
     rack-session (2.1.1)
@@ -326,6 +335,7 @@ GEM
       websocket-extensions (>= 0.1.0)
     websocket-extensions (0.1.5)
     zeitwerk (2.6.18)
+    zlib (3.2.3)
 
 PLATFORMS
   aarch64-linux-gnu
@@ -345,6 +355,7 @@ DEPENDENCIES
   ostruct
   pg
   pry
+  rabbit_messaging!
   rake
   rspec
   rubocop-config-umbrellio
diff --git a/lib/table_sync.rb b/lib/table_sync.rb
@@ -52,6 +52,7 @@ def sync(object_class, **options)
         if_condition: options[:if],
         unless_condition: options[:unless],
         debounce_time: options[:debounce_time],
+        compress: options.fetch(:compress, false)
       ).register_callbacks
     end
 
diff --git a/lib/table_sync/instrument_adapter/active_support.rb b/lib/table_sync/instrument_adapter/active_support.rb
@@ -4,13 +4,14 @@ module TableSync::InstrumentAdapter
   module ActiveSupport
     module_function
 
-    def notify(table:, schema:, event:, direction:, count: 1)
+    def notify(table:, schema:, event:, direction:, count: 1, compress: false)
       ::ActiveSupport::Notifications.instrument "tablesync.#{direction}.#{event}",
                                                 count:,
                                                 table: table.to_s,
                                                 schema: schema.to_s,
                                                 event:,
-                                                direction:
+                                                direction:,
+                                                compress:
     end
   end
 end
diff --git a/lib/table_sync/publishing/batch.rb b/lib/table_sync/publishing/batch.rb
@@ -6,7 +6,8 @@ class TableSync::Publishing::Batch
                 :custom_version,
                 :routing_key,
                 :headers,
-                :event
+                :event,
+                :compress
 
   def initialize(attrs = {})
     attrs = attrs.with_indifferent_access
@@ -17,6 +18,7 @@ def initialize(attrs = {})
     self.routing_key          = attrs[:routing_key]
     self.headers              = attrs[:headers]
     self.event                = attrs.fetch(:event, :update).to_sym
+    self.compress             = attrs.fetch(:compress, false)
 
     validate_required_attributes!
   end
@@ -55,6 +57,7 @@ def attributes
       routing_key: routing_key,
       headers: headers,
       event: event,
+      compress: compress,
     }
   end
 
diff --git a/lib/table_sync/publishing/message/base.rb b/lib/table_sync/publishing/message/base.rb
@@ -5,7 +5,8 @@ class Base
     attr_accessor :custom_version,
                   :object_class,
                   :original_attributes,
-                  :event
+                  :event,
+                  :compress
 
     attr_reader :objects
 
@@ -14,6 +15,7 @@ def initialize(params = {})
       self.object_class          = params[:object_class]
       self.original_attributes   = params[:original_attributes]
       self.event                 = params[:event].to_sym
+      self.compress              = params.fetch(:compress, false)
 
       @objects = find_or_init_objects
 
@@ -65,6 +67,7 @@ def notify!
         table: model_naming.table,
         schema: model_naming.schema,
         event:,
+        compress:,
         direction: :publish,
         count: objects.count,
       )
diff --git a/lib/table_sync/publishing/message/batch.rb b/lib/table_sync/publishing/message/batch.rb
@@ -13,7 +13,7 @@ def initialize(params = {})
 
     def params
       TableSync::Publishing::Params::Batch.new(
-        { object_class:, headers:, routing_key: }.compact,
+        { object_class:, headers:, routing_key:, compress: }.compact,
       ).construct
     end
   end
diff --git a/lib/table_sync/publishing/message/raw.rb b/lib/table_sync/publishing/message/raw.rb
@@ -9,7 +9,8 @@ class Raw
                   :routing_key,
                   :headers,
                   :custom_version,
-                  :event
+                  :event,
+                  :compress
 
     def initialize(params = {})
       self.model_name          = params[:model_name]
@@ -20,6 +21,7 @@ def initialize(params = {})
       self.headers             = params[:headers]
       self.custom_version      = params[:custom_version]
       self.event               = params[:event]
+      self.compress            = params.fetch(:compress, false)
     end
 
     def publish
@@ -35,6 +37,7 @@ def notify!
         table: table_name,
         schema: schema_name,
         event:,
+        compress: compress,
         count: original_attributes.count,
         direction: :publish,
       )
@@ -57,7 +60,7 @@ def data
 
     def params
       TableSync::Publishing::Params::Raw.new(
-        { model_name:, headers:, routing_key: }.compact,
+        { model_name:, headers:, routing_key:, compress: }.compact,
       ).construct
     end
   end
diff --git a/lib/table_sync/publishing/message/single.rb b/lib/table_sync/publishing/message/single.rb
@@ -7,7 +7,7 @@ def object
     end
 
     def params
-      TableSync::Publishing::Params::Single.new(object:).construct
+      TableSync::Publishing::Params::Single.new(object:, compress:).construct
     end
   end
 end
diff --git a/lib/table_sync/publishing/params/base.rb b/lib/table_sync/publishing/params/base.rb
@@ -13,9 +13,12 @@ def construct
         routing_key:,
         headers:,
         exchange_name:,
+        compress: compress,
       )
     end
 
+    attr_accessor :compress
+
     private
 
     def calculated_routing_key
diff --git a/lib/table_sync/publishing/params/batch.rb b/lib/table_sync/publishing/params/batch.rb
@@ -10,6 +10,7 @@ def initialize(attrs = {})
       @exchange_name      = attrs[:exchange_name]
       @routing_key        = attrs[:routing_key]
       @headers            = attrs[:headers]
+      @compress           = attrs.fetch(:compress, false)
     end
 
     def exchange_name
diff --git a/lib/table_sync/publishing/params/single.rb b/lib/table_sync/publishing/params/single.rb
@@ -4,10 +4,11 @@ module TableSync::Publishing::Params
   class Single < Base
     attr_reader :object, :routing_key, :headers
 
-    def initialize(object:)
+    def initialize(object:, compress: false)
       @object      = object
       @routing_key = calculated_routing_key
       @headers     = calculated_headers
+      @compress    = compress
     end
 
     private
diff --git a/lib/table_sync/publishing/raw.rb b/lib/table_sync/publishing/raw.rb
@@ -10,7 +10,8 @@ class TableSync::Publishing::Raw
                 :custom_version,
                 :routing_key,
                 :headers,
-                :event
+                :event,
+                :compress
 
   def initialize(attributes = {})
     attributes = attributes.with_indifferent_access
@@ -23,6 +24,7 @@ def initialize(attributes = {})
     self.routing_key          = attributes[:routing_key]
     self.headers              = attributes[:headers]
     self.event                = attributes.fetch(:event, :update).to_sym
+    self.compress             = attributes.fetch(:compress, false)
   end
 
   require_attributes :model_name, :original_attributes
@@ -47,6 +49,7 @@ def attributes
       routing_key: routing_key,
       headers: headers,
       event: event,
+      compress: compress,
     }
   end
 end
diff --git a/lib/table_sync/publishing/single.rb b/lib/table_sync/publishing/single.rb
@@ -7,14 +7,16 @@ class TableSync::Publishing::Single
                 :original_attributes,
                 :debounce_time,
                 :custom_version,
-                :event
+                :event,
+                :compress
 
   def initialize(attrs = {})
     attrs = attrs.with_indifferent_access
 
     self.object_class         = attrs[:object_class]
     self.original_attributes  = attrs[:original_attributes]
     self.debounce_time        = attrs[:debounce_time]
+    self.compress             = attrs.fetch(:compress, false)
     self.custom_version       = attrs[:custom_version]
     self.event                = attrs.fetch(:event, :update)
   end
@@ -56,6 +58,7 @@ def attributes
       debounce_time: debounce_time,
       custom_version: custom_version,
       event: event,
+      compress: compress,
     }
   end
 
diff --git a/lib/table_sync/setup/active_record.rb b/lib/table_sync/setup/active_record.rb
@@ -16,6 +16,7 @@ def define_after_commit(event)
             original_attributes: attributes,
             event:,
             debounce_time: options[:debounce_time],
+            compress: options.fetch(:compress, false)
           ).publish_later
         end
       end
diff --git a/lib/table_sync/setup/base.rb b/lib/table_sync/setup/base.rb
@@ -6,7 +6,7 @@ class Base
     INVALID_EVENT     = Class.new(StandardError)
     INVALID_CONDITION = Class.new(StandardError)
 
-    attr_accessor :object_class, :debounce_time, :on, :if_condition, :unless_condition
+    attr_accessor :object_class, :debounce_time, :on, :if_condition, :unless_condition, :compress
 
     def initialize(attrs = {})
       attrs.each do |key, value|
@@ -57,6 +57,7 @@ def options_exposed_for_block
         if: if_condition,
         unless: unless_condition,
         debounce_time:,
+        compress: compress,
       }
     end
   end
diff --git a/lib/table_sync/setup/sequel.rb b/lib/table_sync/setup/sequel.rb
@@ -15,6 +15,7 @@ def define_after_commit(event)
               original_attributes: values,
               event:,
               debounce_time: options[:debounce_time],
+              compress: options.fetch(:compress, false),
             ).publish_later
           end
         end
diff --git a/lib/table_sync/version.rb b/lib/table_sync/version.rb
@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module TableSync
-  VERSION = "6.10.0"
+  VERSION = "6.11.0"
 end
diff --git a/spec/publishing/batch_spec.rb b/spec/publishing/batch_spec.rb
@@ -11,6 +11,7 @@
   let(:routing_key)					  { object_class.tableize }
   let(:expected_routing_key) { "a_record_users" }
   let(:headers) { { klass: object_class } }
+  let(:compress) { false }
 
   let(:attributes) do
     {
@@ -20,6 +21,7 @@
       headers:,
       routing_key:,
       custom_version: nil,
+      compress:,
     }
   end
 
@@ -29,6 +31,16 @@
                   TableSync::Publishing::Batch,
                   %i[object_class original_attributes]
 
+  context "when compress option has been provided" do
+    let(:compress) { true }
+
+    it_behaves_like "publisher#publish_now with stubbed message",
+                    TableSync::Publishing::Message::Batch
+    it_behaves_like "publisher#new without expected fields",
+                    TableSync::Publishing::Batch,
+                    %i[object_class original_attributes]
+  end
+
   context "real user" do
     context "sequel" do
       let(:object_class) { "SequelUser" }
diff --git a/spec/publishing/message/batch_spec.rb b/spec/publishing/message/batch_spec.rb
@@ -40,6 +40,7 @@
           object_class: attributes[:object_class],
           routing_key: attributes[:routing_key],
           headers: attributes[:headers],
+          compress: false,
         }
       end
 
@@ -66,6 +67,21 @@
 
         described_class.new(attributes).publish
       end
+
+      context "when compress option has been specified" do
+        let(:attributes) { super().merge(compress: true) }
+        let(:params_attributes) { super().merge(compress: true) }
+
+        it "calls data and params with correct attrs" do
+          expect(data_class).to receive(:new).with(data_attributes)
+          expect(params_class).to receive(:new).with(params_attributes)
+
+          expect(data).to receive(:construct)
+          expect(params).to receive(:construct)
+
+          described_class.new(attributes).publish
+        end
+      end
     end
 
     it "calls Rabbit#publish" do
diff --git a/spec/publishing/message/raw_spec.rb b/spec/publishing/message/raw_spec.rb
diff --git a/spec/publishing/message/single_spec.rb b/spec/publishing/message/single_spec.rb
diff --git a/spec/publishing/params/batch_spec.rb b/spec/publishing/params/batch_spec.rb
diff --git a/spec/publishing/params/raw_spec.rb b/spec/publishing/params/raw_spec.rb
diff --git a/spec/publishing/params/single_spec.rb b/spec/publishing/params/single_spec.rb
diff --git a/spec/publishing/raw_spec.rb b/spec/publishing/raw_spec.rb
diff --git a/spec/publishing/single_spec.rb b/spec/publishing/single_spec.rb
diff --git a/table_sync.gemspec b/table_sync.gemspec
