[Feature] Implement Payload Codec interface and integrate it into the DataConverter (#50)

Author: antstorm

lib/temporal/data_converter.rb

@@ -1,40 +1,73 @@
+require 'temporal/api/common/v1/message_pb'
+require 'temporal/errors'
 require 'temporal/payload_converter'
 
 module Temporal
   class DataConverter
-    def initialize(payload_converter: Temporal::PayloadConverter::DEFAULT)
+    class MissingPayload < Temporal::Error; end
+
+    def initialize(payload_converter: Temporal::PayloadConverter::DEFAULT, payload_codecs: [])
       @payload_converter = payload_converter
+      @payload_codecs = payload_codecs
     end
 
     def to_payloads(data)
       return if data.nil? || Array(data).empty?
 
       payloads = Array(data).map { |value| to_payload(value) }
-      Temporal::Api::Common::V1::Payloads.new(payloads: payloads)
+      Temporal::Api::Common::V1::Payloads.new(payloads: encode(payloads))
     end
 
     def to_payload_map(data)
-      data.to_h { |key, value| [key.to_s, to_payload(value)] }
+      data.to_h do |key, value|
+        payload = to_payload(value)
+        encoded_payload = encode([payload]).first
+        raise MissingPayload, 'Payload Codecs returned no payloads' unless encoded_payload
+
+        [key.to_s, encoded_payload]
+      end
     end
 
     def from_payloads(payloads)
       return unless payloads
 
-      payloads.payloads.map { |payload| from_payload(payload) }
+      decode(payloads.payloads)
+        .map { |payload| from_payload(payload) }
     end
 
     def from_payload_map(payload_map)
       return unless payload_map
 
       # Protobuf's Hash isn't compatible with the native Hash, ignore rubocop here
-      # rubocop:disable Style/HashTransformValues, Style/MapToHash
-      payload_map.map { |key, value| [key, from_payload(value)] }.to_h
-      # rubocop:enable Style/HashTransformValues, Style/MapToHash
+      # rubocop:disable Style/MapToHash
+      payload_map.map do |key, payload|
+        decoded_payload = decode([payload]).first
+        raise MissingPayload, 'Payload Codecs returned no payloads' unless decoded_payload
+
+        [key, from_payload(decoded_payload)]
+      end.to_h
+      # rubocop:enable Style/MapToHash
     end
 
     private
 
-    attr_reader :payload_converter
+    attr_reader :payload_converter, :payload_codecs
+
+    def encode(payloads)
+      payload_codecs.each do |codec|
+        payloads = codec.encode(payloads)
+      end
+
+      payloads
+    end
+
+    def decode(payloads)
+      payload_codecs.reverse_each do |codec|
+        payloads = codec.decode(payloads)
+      end
+
+      payloads
+    end
 
     def to_payload(data)
       payload_converter.to_payload(data)

lib/temporal/payload_codec/base.rb

@@ -0,0 +1,13 @@
+module Temporal
+  module PayloadCodec
+    class Base
+      def encode(_payloads)
+        raise NoMethodError, 'must implement #encode'
+      end
+
+      def decode(_payloads)
+        raise NoMethodError, 'must implement #decode'
+      end
+    end
+  end
+end

sig/temporal/data_converter.rbs

@@ -1,6 +1,12 @@
 module Temporal
   class DataConverter
-    def initialize: (?payload_converter: Temporal::PayloadConverter::_PayloadConverter) -> void
+    class MissingPayload < Temporal::Error
+    end
+
+    def initialize: (
+      ?payload_converter: Temporal::PayloadConverter::_PayloadConverter,
+      ?payload_codecs: Array[Temporal::PayloadCodec::_PayloadCodec],
+    ) -> void
     def to_payloads: (Array[untyped]? data) -> Temporal::Api::Common::V1::Payloads?
     def to_payload_map: (Hash[String | Symbol, untyped] data) -> Hash[String, Temporal::Api::Common::V1::Payload]
     def from_payloads: (Temporal::Api::Common::V1::Payloads? payloads) -> Array[untyped]?
@@ -9,7 +15,10 @@ module Temporal
     private
 
     attr_reader payload_converter: Temporal::PayloadConverter::_PayloadConverter
+    attr_reader payload_codecs: Array[Temporal::PayloadCodec::_PayloadCodec]
 
+    def encode: (Array[Temporal::Api::Common::V1::Payload]) -> Array[Temporal::Api::Common::V1::Payload]
+    def decode: (Array[Temporal::Api::Common::V1::Payload]) -> Array[Temporal::Api::Common::V1::Payload]
     def to_payload: (untyped data) -> Temporal::Api::Common::V1::Payload
     def from_payload: (Temporal::Api::Common::V1::Payload payload) -> untyped
   end

sig/temporal/payload_codec/base.rbs

@@ -0,0 +1,12 @@
+module Temporal
+  module PayloadCodec
+    interface _PayloadCodec
+      def encode: (Array[Temporal::Api::Common::V1::Payload]) -> Array[Temporal::Api::Common::V1::Payload]
+      def decode: (Array[Temporal::Api::Common::V1::Payload]) -> Array[Temporal::Api::Common::V1::Payload]
+    end
+
+    class Base
+      include _PayloadCodec
+    end
+  end
+end

spec/unit/temporal/data_converter_spec.rb

@@ -1,9 +1,79 @@
+require 'temporal/api/common/v1/message_pb'
 require 'temporal/data_converter'
 require 'temporal/payload_converter'
+require 'temporal/payload_codec/base'
+
+class TestConcatenatingPayloadCodec < Temporal::PayloadCodec::Base
+  ENCODING = 'mixed'.freeze
+
+  def initialize(separator)
+    super()
+    @separator = separator
+  end
+
+  def encode(payloads)
+    data = payloads.map do |payload|
+      Temporal::Api::Common::V1::Payload.encode(payload)
+    end.join(separator)
+
+    return [] if data.empty?
+
+    [Temporal::Api::Common::V1::Payload.new(
+      metadata: { encoding: ENCODING },
+      data: data
+    )]
+  end
+
+  def decode(payloads)
+    return if payloads.empty?
+    raise 'unexpected number of payloads' if payloads.length > 1
+
+    payloads.first.data.split(separator).map do |bytes|
+      Temporal::Api::Common::V1::Payload.decode(bytes)
+    end
+  end
+
+  private
+
+  attr_reader :separator
+end
+
+class TestFaultyPayloadCodec < Temporal::PayloadCodec::Base
+  def encode(_payloads)
+    []
+  end
+
+  def decode(_payloads)
+    []
+  end
+end
+
+# This codec doesn't do much, but it ensures that codecs are applied in the correct order
+class TestEncodingSwappingPayloadCodec < Temporal::PayloadCodec::Base
+  FROM_ENCODING = TestConcatenatingPayloadCodec::ENCODING
+  TO_ENCODING = 'swapped'.freeze
+
+  def encode(payloads)
+    payload = payloads.first
+    raise 'unexpected payload' unless payload.metadata['encoding'] == FROM_ENCODING
+
+    payload.metadata['encoding'] = TO_ENCODING
+    [payload]
+  end
+
+  def decode(payloads)
+    payload = payloads.first
+    raise 'unexpected payload' unless payload.metadata['encoding'] == TO_ENCODING
+
+    payload.metadata['encoding'] = FROM_ENCODING
+    [payload]
+  end
+end
 
 describe Temporal::DataConverter do
-  subject { described_class.new(payload_converter: converter) }
+  subject { described_class.new(payload_converter: converter, payload_codecs: codecs) }
   let(:converter) { Temporal::PayloadConverter::DEFAULT }
+  let(:codecs) { [] }
   let(:nil_payload) do
     Temporal::Api::Common::V1::Payload.new(
       metadata: { 'encoding' => Temporal::PayloadConverter::Nil::ENCODING },
@@ -15,10 +85,15 @@
       data: '"test"',
     )
   end
+  let(:test_codec) { TestConcatenatingPayloadCodec.new('$$$') }
+  let(:faulty_codec) { TestFaultyPayloadCodec.new }
+  let(:swap_codec) { TestEncodingSwappingPayloadCodec.new }
 
   before do
     allow(converter).to receive(:to_payload).and_call_original
     allow(converter).to receive(:from_payload).and_call_original
+    allow(test_codec).to receive(:encode).and_call_original
+    allow(test_codec).to receive(:decode).and_call_original
   end
 
   describe '#to_payloads' do
@@ -45,6 +120,21 @@
       expect(converter).to have_received(:to_payload).with('test').once
       expect(converter).to have_received(:to_payload).with(nil).once
     end
+
+    context 'with payload codecs' do
+      let(:codecs) { [test_codec] }
+
+      it 'encodes the payloads' do
+        result = subject.to_payloads(['test', nil])
+
+        expect(result).to be_a(Temporal::Api::Common::V1::Payloads)
+        expect(result.payloads.length).to eq(1)
+        expect(result.payloads.first.metadata['encoding'])
+          .to eq(TestConcatenatingPayloadCodec::ENCODING)
+
+        expect(test_codec).to have_received(:encode).once
+      end
+    end
   end
 
   describe '#to_payload_map' do
@@ -64,6 +154,30 @@
       expect(converter).to have_received(:to_payload).with('test').once
       expect(converter).to have_received(:to_payload).with(nil).once
     end
+
+    context 'with payload codecs' do
+      let(:codecs) { [test_codec] }
+
+      it 'encodes each payload' do
+        result = subject.to_payload_map({ 'one' => 'test', 'two' => nil })
+
+        expect(result.length).to eq(2)
+        expect(result['one'].metadata['encoding']).to eq(TestConcatenatingPayloadCodec::ENCODING)
+        expect(result['two'].metadata['encoding']).to eq(TestConcatenatingPayloadCodec::ENCODING)
+
+        expect(test_codec).to have_received(:encode).twice
+      end
+    end
+
+    context 'with a faulty codec' do
+      let(:codecs) { [test_codec, faulty_codec] }
+
+      it 'raises an error' do
+        expect do
+          subject.to_payload_map({ 'one' => 'test', 'two' => nil })
+        end.to raise_error(described_class::MissingPayload, 'Payload Codecs returned no payloads')
+      end
+    end
   end
 
   describe '#from_payloads' do
@@ -79,9 +193,26 @@
       expect(converter).to have_received(:from_payload).with(json_payload).once
       expect(converter).to have_received(:from_payload).with(nil_payload).once
     end
+
+    context 'with payload codecs' do
+      let(:codecs) { [test_codec] }
+
+      it 'decodecs the payloads' do
+        mixed_payloads = test_codec.encode([json_payload, nil_payload])
+        payloads = Temporal::Api::Common::V1::Payloads.new(payloads: mixed_payloads)
+
+        result = subject.from_payloads(payloads)
+
+        expect(result).to eq(['test', nil])
+        expect(test_codec).to have_received(:decode).once
+      end
+    end
   end
 
   describe '#from_payload_map' do
+    let(:json_encoded) { test_codec.encode([json_payload]).first }
+    let(:nil_encoded) { test_codec.encode([nil_payload]).first }
+
     it 'returns nil when nothing is given' do
       expect(subject.from_payload_map(nil)).to eq(nil)
     end
@@ -93,6 +224,27 @@
       expect(converter).to have_received(:from_payload).with(json_payload).once
       expect(converter).to have_received(:from_payload).with(nil_payload).once
     end
+
+    context 'with payload codecs' do
+      let(:codecs) { [test_codec] }
+
+      it 'decodecs each payload' do
+        result = subject.from_payload_map({ 'one' => json_encoded, 'two' => nil_encoded })
+
+        expect(result).to eq({ 'one' => 'test', 'two' => nil })
+        expect(test_codec).to have_received(:decode).twice
+      end
+    end
+
+    context 'with a faulty codec' do
+      let(:codecs) { [faulty_codec, test_codec] }
+
+      it 'raises an error' do
+        expect do
+          subject.from_payload_map({ 'one' => json_encoded, 'two' => nil_encoded })
+        end.to raise_error(described_class::MissingPayload, 'Payload Codecs returned no payloads')
+      end
+    end
   end
 
   describe 'full circle' do
@@ -113,5 +265,21 @@
 
       expect(subject.from_payload_map(subject.to_payload_map(input))).to eq(input)
     end
+
+    context 'with payload codecs' do
+      let(:codecs) { [test_codec, swap_codec] }
+
+      it 'converts values to payloads and back' do
+        input = ['test', nil]
+
+        expect(subject.from_payloads(subject.to_payloads(input))).to eq(input)
+      end
+
+      it 'converts values map to payloads map and back' do
+        input = { 'one' => 'test', 'two' => nil }
+
+        expect(subject.from_payload_map(subject.to_payload_map(input))).to eq(input)
+      end
+    end
   end
 end