Fix sending and receiving of custom headers.

Author: jemc

lib/rabbitmq/ffi/ext.rb

@@ -4,6 +4,7 @@
 
 require_relative 'ext/connection_info'
 require_relative 'ext/field_value'
+require_relative 'ext/array'
 require_relative 'ext/table'
 
 require_relative 'ext/method'

lib/rabbitmq/ffi/ext/array.rb

@@ -0,0 +1,46 @@
+
+module RabbitMQ
+  module FFI
+    
+    class Array
+      include Enumerable
+      
+      def each(*a, &b)
+        entry_ptr = self[:entries]
+        entries   = self[:num_entries].times.map do |i|
+          FFI::FieldValue.new(entry_ptr + i * FFI::FieldValue.size)
+        end
+        entries.each(*a, &b)
+      end
+      
+      def to_a(free=false)
+        result = self.map do |item|
+          item.to_value(free)
+        end
+        
+        clear if free
+        result
+      end
+      
+      def free!
+        self.each(&:free!)
+        FFI.free(self[:entries])
+        clear
+      end
+      
+      def self.from_a(items)
+        size      = items.size
+        entry_ptr = Util.mem_ptr(size * FFI::FieldValue.size, release: false)
+        items.each_with_index do |item, idx|
+          FFI::FieldValue.new(entry_ptr + idx * FFI::FieldValue.size).apply(item)
+        end
+        
+        obj = new
+        obj[:num_entries] = size
+        obj[:entries]     = entry_ptr
+        obj
+      end
+    end
+    
+  end
+end

lib/rabbitmq/ffi/ext/field_value.rb

@@ -19,8 +19,7 @@ def to_value(free=false)
         when :utf8;      value.to_s(free).force_encoding(Encoding::UTF_8)
         when :timestamp; Time.at(value / 1000.0)
         when :table;     value.to_h(free)
-        when :array;     value.to_array_not_yet_implemented!
-        when :decimal;   value.to_value_not_yet_implemented!
+        when :array;     value.to_a(free)
         else value
         end
 
@@ -29,19 +28,30 @@ def to_value(free=false)
       end
 
       def free!
-        kind   = self[:kind]
-        value  = self[:value][value_member(kind)]
+        kind  = self[:kind]
+        value = self[:value][value_member(kind)]
         value.free! if value.respond_to? :free!
-        clear
+        self
       end
 
-      def self.from(value)
-        obj = new
-        obj[:kind], obj[:value] = case value
-        when String; [:bytes, FieldValueValue.new(Bytes.from_s(value).pointer)]
-        else raise NotImplementedError
+      def apply(value)
+        self[:kind], self[:value] = case value
+        when ::String; [:bytes, FieldValueValue.new(Bytes.from_s(value).pointer)]
+        when ::Symbol; [:bytes, FieldValueValue.new(Bytes.from_s(value.to_s).pointer)]
+        when ::Array;  [:array, FieldValueValue.new(Array.from_a(value).pointer)]
+        when ::Hash;   [:table, FieldValueValue.new(Table.from(value).pointer)]
+        when ::Fixnum; [:i64,       (v=FieldValueValue.new; v[:i64]=value; v)]
+        when ::Float;  [:f64,       (v=FieldValueValue.new; v[:f64]=value; v)]
+        when ::Time;   [:timestamp, (v=FieldValueValue.new; v[:u64]=value.to_i*1000; v)]
+        when true;     [:boolean,   (v=FieldValueValue.new; v[:boolean]=true; v)]
+        when false;    [:boolean,   (v=FieldValueValue.new; v[:boolean]=false; v)]
+        else raise NotImplementedError, "#{self.class}.from(#<#{value.class}>)"
         end
-        obj
+        self
+      end
+      
+      def self.from(value)
+        new.apply(value)
       end
     end
 

spec/channel_spec.rb

@@ -274,4 +274,47 @@
     subject.queue_delete("my_queue")
   end
 
+  it "can send and receive messages with custom headers" do
+    subject.queue_delete("my_queue")
+    subject.queue_declare("my_queue")
+    time = Time.now
+    
+    res = subject.basic_publish("message_body", "", "my_queue",
+      headers: { "x-my-header" => [
+        {
+          "foo" => 88,
+          "bar" => 88.8,
+          "baz" => "string1",
+          "time" => (time + 1),
+          "bool" => true
+        },
+        {
+          "foo" => 99,
+          "bar" => 99.9,
+          "baz" => "string2",
+          "time" => (time - 1),
+          "bool" => false
+        },
+      ]}
+    )
+    res.should eq true
+    
+    res = subject.basic_get("my_queue", no_ack: true)
+    headers = res[:header][:headers]
+    headers.should be_a Hash
+    headers["x-my-header"][0]["foo"].should eq 88
+    headers["x-my-header"][1]["foo"].should eq 99
+    headers["x-my-header"][0]["bar"].should eq 88.8
+    headers["x-my-header"][1]["bar"].should eq 99.9
+    headers["x-my-header"][0]["baz"].should eq "string1"
+    headers["x-my-header"][1]["baz"].should eq "string2"
+    headers["x-my-header"][0]["time"].to_i.should eq((time + 1).to_i)
+    headers["x-my-header"][1]["time"].to_i.should eq((time - 1).to_i)
+    headers["x-my-header"][0]["bool"].should eq true
+    headers["x-my-header"][1]["bool"].should eq false
+    
+    res = subject.basic_get("my_queue", no_ack: true)
+    res.has_key?(:header).should_not be
+  end
+  
 end