Class: FeatBit::WebSocketDataSynchronizer

Inherits:
Object
  • Object
show all
Defined in:
lib/featbit/web_socket_data_synchronizer.rb

Constant Summary collapse

ALPHABETS = 
{ "0" => "Q", "1" => "B", "2" => "W", "3" => "S", "4" => "P",
"5" => "H", "6" => "D", "7" => "X", "8" => "Z", "9" => "U" }.freeze
PING_INTERVAL = 
10.0

Instance Method Summary collapse

Constructor Details

#initialize(options:, data_store:, status_provider:, on_flags_changed: nil, connector: nil) ⇒ WebSocketDataSynchronizer

Returns a new instance of WebSocketDataSynchronizer.



15
16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/featbit/web_socket_data_synchronizer.rb', line 15

def initialize(options:, data_store:, status_provider:, on_flags_changed: nil, connector: nil)
  @options = options
  @data_store = data_store
  @status_provider = status_provider
  @on_flags_changed = on_flags_changed
  @connector = connector || method(:connect)
  @closed = false
  @socket_mutex = Mutex.new
  @lifecycle_mutex = Mutex.new
  @thread = nil
  @close_result = nil
end

Instance Method Details

#close ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/featbit/web_socket_data_synchronizer.rb', line 41

def close
  @lifecycle_mutex.synchronize do
    return @close_result unless @close_result.nil?

    @closed = true
    socket = @socket_mutex.synchronize { @socket }
    socket_closed = safe_close_socket(socket)
    @thread&.join(5) unless Thread.current.equal?(@thread)
    @close_result = socket_closed && !@thread&.alive?
  end
rescue StandardError => e
  @options.logger&.warn("FeatBit synchronizer close failed: #{e.message}")
  false
end

#process_message(message) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/featbit/web_socket_data_synchronizer.rb', line 56

def process_message(message)
  envelope = message.is_a?(String) ? JSON.parse(message) : message
  return false unless fetch(envelope, "messageType") == "data-sync"

  data = fetch(envelope, "data", {})
  event_type = fetch(data, "eventType")
  return false unless valid_data?(data, event_type)

  old_keys = @data_store.all_flags.keys
  if event_type == "full"
    changed = @data_store.init(data)
    changed_keys = changed ? old_keys | @data_store.all_flags.keys : []
  else
    changed, changed_keys = process_patch(data)
  end

  changed_keys.each { |key| safely_notify(key) }
  return false unless changed

  @status_provider.update(Status::READY)
  true
rescue JSON::ParserError => e
  @status_provider.update(Status::FAILED, message: "invalid data: #{e.message}")
  false
rescue StandardError => e
  fail_status(e)
  false
end

#start ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/featbit/web_socket_data_synchronizer.rb', line 28

def start
  @lifecycle_mutex.synchronize do
    return false if @closed || @thread&.alive?

    @thread = Thread.new { run }
    @thread.name = "featbit-websocket-sync" if @thread.respond_to?(:name=)
    true
  end
rescue StandardError => e
  fail_status(e)
  false
end