class Rage::Cable::Protocols::Base

def broadcast(name, data)
  @subscription_identifiers[name].each do |params|
    ::Iodine.publish("cable:#{name}:#{stream_id(params)}", serialize(params, data))
  end
end

def init(router)
  @router = router
  # Hash<String(stream name) => Set<Hash>(subscription params)>
  @subscription_identifiers = Hash.new { |hash, key| hash[key] = Set.new }
  Iodine.on_state(:pre_start) do
    # this is a fallback to synchronize subscription identifiers across different worker processes;
    # we expect connections to be distributed among all workers, so this code will almost never be called;
    # we also synchronize subscriptions with the master process so that the forks that are spun up instead
    # of the crashed ones also had access to the identifiers;
    Iodine.subscribe("cable:synchronize") do |_, subscription_msg|
      stream_name, params = Rage::ParamsParser.json_parse(subscription_msg)
      @subscription_identifiers[stream_name] << params
    end
  end
  Iodine.on_state(:on_finish) do
    Iodine.unsubscribe("cable:synchronize")
  end
end

def protocol_definition
  HANDSHAKE_HEADERS
end

def stream_id(params)
  Digest::MD5.hexdigest(params.to_s)
end

def subscribe(connection, name, params)
  connection.subscribe("cable:#{name}:#{stream_id(params)}")
  unless @subscription_identifiers[name].include?(params)
    @subscription_identifiers[name] << params
    ::Iodine.publish("cable:synchronize", [name, params].to_json)
  end
end

def supports_rpc?
  true
end