[FEATURE] Implement Kafka-specific Create command

The command sends message(s) to the Kafka broker(s).
The method `#using` allows reloading the command by
updating its dataset.

== Added

* ROM::Kafka::Commands (the namespace)
* ROM::Kafka::Commands::Create

Author: nepalez

config/metrics/flay.yml

@@ -1,2 +1,2 @@
 ---
-minimum_score: 8
+minimum_score: 9

lib/rom/kafka.rb

@@ -19,6 +19,7 @@ module Kafka
     require_relative "kafka/dataset"
     require_relative "kafka/gateway"
     require_relative "kafka/relation"
+    require_relative "kafka/create"
 
   end # module Kafka
 

lib/rom/kafka/create.rb

@@ -0,0 +1,55 @@
+# encoding: utf-8
+
+module ROM::Kafka
+
+  # The namespace for Kafka-specific ROM commands
+  #
+  module Commands
+
+    # The Kafka-specific implementation of ROM::Commands::Create
+    #
+    # @example
+    #   ROM.use(:auto_registration)
+    #   ROM.setup(:kafka, :consumer, "localhost:9092")
+    #
+    #   class Users < ROM::Relation[:kafka]
+    #     dataset "log.users"
+    #   end
+    #
+    #   class AddUsers < ROM::Commands::Create[:kafka]
+    #   end
+    #
+    #   rom = ROM.finalize.env
+    #   rom.relation(:users).using(max_wait_ms: 100).offset(10).to_a
+    #   # => [{ value: "something", topic: "log", key: "users", offset: 10 }]
+    #
+    class Create < ROM::Commands::Create
+
+      adapter :kafka
+
+      # Sends tuples to the current topic/partition of Kafka
+      #
+      # @param (see ROM::Kafka::Dataset#publish)
+      #
+      # @return (see ROM::Kafka::Dataset#publish)
+      #
+      def execute(*tuples)
+        dataset.publish(*tuples)
+      end
+
+      # Returns the new commmand where relation's dataset is updated
+      # using given options
+      #
+      # @param [Hash] options
+      #
+      # @return [ROM::Kafka::Commands::Create]
+      #
+      def using(options)
+        new relation.using(options)
+      end
+
+    end # class Create
+
+  end # module Commands
+
+end # module ROM::Kafka

spec/unit/create_spec.rb

@@ -0,0 +1,57 @@
+# encoding: utf-8
+
+describe ROM::Kafka::Commands::Create do
+
+  let(:command)  { described_class.new relation }
+  let(:relation) { double :relation, using: updated, dataset: dataset }
+  let(:updated)  { double :updated }
+  let(:dataset)  { double :dataset, publish: tuples }
+  let(:tuples)   { [double(:first_message), double(:second_message)] }
+
+  describe ".adapter" do
+    subject { described_class.adapter }
+
+    it { is_expected.to eql(:kafka) }
+  end # describe .adapter
+
+  describe ".new" do
+    subject { command }
+
+    it { is_expected.to be_kind_of ROM::Commands::Create }
+  end # describe .new
+
+  describe "#relation" do
+    subject { command.relation }
+
+    it { is_expected.to eql(relation) }
+  end # describe #relation
+
+  describe "#using" do
+    subject { command.using(options) }
+
+    let(:options) { { foo: :FOO, bar: :BAR } }
+
+    it "returns a relation" do
+      expect(subject).to be_kind_of described_class
+    end
+
+    it "updates the relation" do
+      expect(relation).to receive(:using).with(options)
+      expect(subject.relation).to eql(updated)
+    end
+  end # describe #using
+
+  describe "#execute" do
+    subject { command.execute(*tuples) }
+
+    it "publishes tuples to the dataset" do
+      expect(dataset).to receive(:publish).with(*tuples)
+      subject
+    end
+
+    it "returns tuples" do
+      expect(subject).to eql(tuples)
+    end
+  end # describe #execute
+
+end # describe ROM::Kafka::Relation