如何测试一个块在线程中被调用？

Question

我正在努力将ruby-mqtt gem 包装到一个实现subscribe 和publish 方法的类中。 subscribe 方法连接到服务器并在单独的线程中侦听，因为此调用是同步的。

module PubSub
class MQTT
    attr_accessor :host, :port, :username, :password

    def initialize(params = {})
        params.each do |attr, value|
            self.public_send("#{attr}=", value)
        end if params
        super()
    end

    def connection_options
        {
            remote_host: self.host,
            remote_port: self.port,
            username: self.username,
            password: self.password,
        }
    end

    def subscribe(name, &block)
        channel = name
        connect_opts = connection_options
        code_block = block
        ::Thread.new do
            ::MQTT::Client.connect(connect_opts) do |c|
                c.get(channel) do |topic, message|
                    puts "channel: #{topic} data: #{message.inspect}"
                    code_block.call topic, message
                end
            end
        end
    end

    def publish(channel = nil, data)
        ::MQTT::Client.connect(connection_options) do |c|
            c.publish(channel, data)
        end
    end     

end
end

我有一个使用 rspec 编写的测试来测试该类，但它没有通过。

mqtt = ::PubSub::MQTT.new({host: "localhost",port: 1883})
block = lambda { |channel, data|  puts "channel: #{channel} data: #{data.inspect}"}
block.should_receive(:call).with("channel", {"some" => "data"})
thr = mqtt.subscribe("channel", &block)
mqtt.publish("channel", {"some" => "data"})

当我运行以下ruby-mqtt-example 时，我现在完全遇到了问题。

uri = URI.parse ENV['CLOUDMQTT_URL'] || 'mqtt://localhost:1883'
conn_opts = {
  remote_host: uri.host,
  remote_port: uri.port,
  username: uri.user,
  password: uri.password,
}

# Subscribe example
Thread.new do
  puts conn_opts
  MQTT::Client.connect(conn_opts) do |c|
    # The block will be called when you messages arrive to the topic
    c.get('test') do |topic, message|
      puts "#{topic}: #{message}"
    end
  end
end

# Publish example
puts conn_opts
MQTT::Client.connect(conn_opts) do |c|
  # publish a message to the topic 'test'
  loop do
    c.publish('test', 'Hello World')
    sleep 1
  end
end

所以我的问题是，当我简单地创建一个类并分离发布和订阅逻辑时，我做错了什么？我的猜测是它与函数调用中的线程有关，但我似乎无法弄清楚。非常感谢任何帮助。

更新

我相信我知道为什么测试没有通过，这是因为当我将 lambda 传递给 subscribe 并期望它接收调用时，它实际上不会在退出方法时或直到 @ 987654331@ 被调用。所以我想把这个问题改写成：我如何测试一个块在一个线程中被调用？如果有人回答“你不知道”，那么问题是：如何测试该块在无限循环中被调用，例如在 ruby-mqtt gem 中调用 get 的示例。

Answer 1

RSpec 期望机制可以很好地处理线程，如下例所示，它通过了：

def foo(&block)
  block.call(42)
end

describe "" do
  it "" do
    l = lambda {}
    expect(l).to receive(:call).with(42)
    Thread.new { foo(&l) }.join
  end
end

join 在继续之前等待线程完成。