# Why doesn't this plugin add messages to the queue?

**URL:** <https://discuss.elastic.co/t/why-doesnt-this-plugin-add-messages-to-the-queue/51229>\
**Category:** Logstash\
**Created:** [May 28, 2016, 4:08pm UTC](https://discuss.elastic.co/t/why-doesnt-this-plugin-add-messages-to-the-queue/51229 "2016-05-28T16:08:44Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![spuder](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/spuder/32/44895_2.png) [@spuder](https://discuss.elastic.co/u/spuder)\
**Post date:** [May 28, 2016, 4:08pm UTC](https://discuss.elastic.co/t/why-doesnt-this-plugin-add-messages-to-the-queue/51229/1 "2016-05-28T16:08:44Z")

</div>

I'm trying to write my first logstash input plugin by [following the guide](https://www.elastic.co/guide/en/logstash/current/_how_to_write_a_logstash_input_plugin.html). It uses the [mqtt ruby gem](https://github.com/njh/ruby-mqtt) to fetch from a mqtt message bus.

The method 'run' is able to fetch from mqtt, and the 'puts' statement shows the message. However the message never makes it to the queue.

`queue << message`

```
./bin/logstash -e 'input { stdin{} mqtt{} } output {stdout { codec => rubydebug }}'

```

```auto
# encoding: utf-8
require "logstash/inputs/base"
require "logstash/namespace"
require "stud/interval"
require "socket" # for Socket.gethostname
require 'mqtt'

class LogStash::Inputs::Mqtt < LogStash::Inputs::Base
  config_name "mqtt"
  default :codec, "plain"
  config :state_topic, :validate => :string, :default => 'test'

  public
  def register

    @client = MQTT::Client.new
    @client.host = 'test.mosquitto.org'
    @host = Socket.gethostname
  end # def register

  def run(queue)
    @client.connect()
    @client.subscribe( 'test' )
    while !stop?
      @client.get do |topic,message|
        puts "#{topic}: #{message}"
        decorate(message)
        queue << message
      end
      Stud.stoppable_sleep(@interval) { stop? }
    end # loop
    @client.disconnect()
  end # def run

  def stop
    @client.disconnect()
  end
end # class LogStash::Inputs::Mqtt

```

Here I typed 'herpderp' into stdin, and logstash received and parsed it correctly.  
I put the message "testing the message bus" from a separate mqtt broker. The message is received by logstash and printed, but is not parsed by logtash.

```auto
herpderp
{
       "message" => "herpderp",
    "@timestamp" => 2016-05-28T16:04:00.338Z,
          "host" => "vagrant-ubuntu-trusty-64",
      "@version" => "1"
}
test: testing the message bus

```

I've compared my 'run' method with many of the other similar logstash plugins. I dont' see why `queue << message` isn't working.

[stdin](https://github.com/logstash-plugins/logstash-input-stdin/blob/master/lib/logstash/inputs/stdin.rb)  
[zeromq](https://github.com/logstash-plugins/logstash-input-zeromq/blob/master/lib/logstash/inputs/zeromq.rb)  
[lumberjack](https://github.com/logstash-plugins/logstash-input-lumberjack/blob/master/lib/logstash/inputs/lumberjack.rb)  
[tcp](https://github.com/logstash-plugins/logstash-input-tcp/blob/master/lib/logstash/inputs/tcp.rb)

---

<div class="post-metadata">

**Author:** ![spuder](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/spuder/32/44895_2.png) [@spuder](https://discuss.elastic.co/u/spuder)\
**Post date:** [May 28, 2016, 9:54pm UTC](https://discuss.elastic.co/t/why-doesnt-this-plugin-add-messages-to-the-queue/51229/2 "2016-05-28T21:54:54Z")

</div>

Sharing a working solution. The message needs to be decoded by the codec first.

```auto
  def run(output_queue)
    @client.connect()
    @client.subscribe( 'test' )
    while !stop?
      @client.get do |topic,message|
        @codec.decode(message) do |event|
          decorate(event)
          event["host"] = @host if !event.include?("host")
          output_queue << event
        end
      end
      Stud.stoppable_sleep(@interval) { stop? }
    end # loop
    @client.disconnect()
  end # def run

```

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 4:55am UTC](https://discuss.elastic.co/t/why-doesnt-this-plugin-add-messages-to-the-queue/51229/3 "2017-07-06T04:55:45Z")

</div>