in_gcloud_pubsub.rb in fluent-plugin-gcloud-pubsub-custom-0.3.4.pre

- old
+ new

@@ -115,10 +115,11 @@
       start_rpc if @enable_rpc
 
       @subscriber = Fluent::GcloudPubSub::Subscriber.new @project, @key, @topic, @subscription
       log.debug "connected subscription:#{@subscription} in project #{@project}"
 
+      @emit_guard = Mutex.new
       @stop_subscribing = false
       @subscribe_threads = []
       @pull_threads.times do
         @subscribe_threads.push Thread.new(&method(:subscribe))
       end
@@ -205,9 +206,14 @@
           else
             raise FailedParseError.new "pattern not match: #{line.inspect}"
           end
         end
       end
-      router.emit_stream(@tag, es)
+
+      # There are some output plugins not to supposed to be called with multi-threading.
+      # Maybe remove in the future.
+      @emit_guard.synchronize do
+        router.emit_stream(@tag, es)
+      end
     end
   end
 end