Subsequent subscribe takes between 0-ping seconds

[Karazum]

When subscribing to a second channel (in the same process) the second subscribe takes between 0 and whatever the ping value in seconds is set.

If a message is sent to that channel before the subscribe is successful, the message will not be received by the subscribed client, but the subscribe will finish when the message is sent.

I have created this repository https://github.com/Karazum/faye_subscribe_test
To test this issue. The faye_server.rb script contains 3 tests.

1st test:

• thread 1 subscribes -> subscribe is successful in less than 0.01 seconds
• after 3 seconds
• thread 2 subscribes to second channel
• thread 2 finishes subscribe in ~9 seconds

2nd test:

• thread 1 subscribes -> subscribe is successful in less than 0.01 seconds
• after 3 seconds
• thread 2 subscribes
• after another 3 seconds, send message to thread 2
• thread 2 finishes subscribe in ~2 seconds, but does NOT receive the message send

3rd test:

• thread 1 subscribes -> subscribe is successful in less than 0.01 seconds
• after 3 seconds
• thread 2 subscribes
• after another 3 seconds, send 2 message to thread 2
• thread 2 finishes subscribe in ~2 seconds, but it only receives 1 message!

To conclude:

• All subsequent subscribes are not instant
• Messages to channel that are not finished to subscribed channels are lost, but subscribe is finished much eariler.

[jcoglan]

This is most likely caused by the use of threads, sleep calls, blocking loops and I/O in this script, as well as running a threaded server and EventMachine in the same process. If I put the server in its own file and run it as a separate process, and write the client code to use EventMachine APIs for timers, everything works as expected:

require "bundler"
require "bundler/setup"
require "rack/handler/puma"
require "net/http"

Bundler.require

server_timeout = 30
$web_socket_url = "http://0.0.0.0:9393/faye"

# Faye.logger = Logger.new(STDOUT)

EM.run {
  client_timeout = server_timeout + 1
  client = Faye::Client.new($web_socket_url, timeout: client_timeout)

  def do_sub(client, i)
    puts "Subscribing to channel #{i}"

    start_time = Time.now.to_f

    sub = client.subscribe("/channel/#{i}").with_channel do |channel, message|
      p ["Got message on channel #{i}", channel, message]
    end

    sub.callback do
      duration = Time.now.to_f - start_time
      puts "Channel #{i} subscribed in #{duration} seconds"
    end
  end

  def send_message(i)
    payload = {
      channel: "/channel/#{i}",
      data: {
        foo: "bar"
      },
    }
    p [:send, payload]
    uri = URI.parse($web_socket_url)
    Net::HTTP.post_form(uri, message: JSON.dump(payload))
  end

  do_sub(client, 1)

  EM.add_timer(3) do
    do_sub(client, 2)
    EM.add_timer(3) { send_message(2) }
  end
}