support init subscription before enqueue message

archfish · Dec 31, 2019 · 384ba0c · 384ba0c
1 parent 8555284
commit 384ba0c
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 2 deletions.
diff --git a/lib/mqjob/version.rb b/lib/mqjob/version.rb
@@ -1,3 +1,3 @@
 module Mqjob
-  VERSION = "0.3.9"
+  VERSION = "0.3.10"
 end
diff --git a/lib/mqjob/worker.rb b/lib/mqjob/worker.rb
@@ -87,10 +87,11 @@ def from_topic(name, opts={})
       # opts
       #   in publish message in X seconds
       #   at publish message at specific time
+      #   init_subscription Boolean 是否先初始化一个订阅
       def enqueue(msg, opts={})
         @mq ||= Plugin.client(topic_opts[:client])
 
-        @mq.publish(topic, msg, topic_opts)
+        @mq.publish(topic, msg, topic_opts.merge(opts))
         true
       end
     end

diff --git a/lib/plugin/pulsar.rb b/lib/plugin/pulsar.rb
@@ -12,7 +12,10 @@ def listen(topic, worker, opts = {})
     # opts
     #   in publish message in X seconds
     #   at publish message at specific time
+    #   init_subscription Boolean
     def publish(topic, msg, opts = {})
+      create_consumer(topic, opts) if opts[:init_subscription]
+
       base_cmd = ::Pulsar::Proto::BaseCommand.new(
         type: ::Pulsar::Proto::BaseCommand::Type::SEND,
         send: ::Pulsar::Proto::CommandSend.new(