Class: LaunchDarkly::EventProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/ldclient-rb/events.rb

Instance Method Summary collapse

Constructor Details

#initialize(sdk_key, config, client = nil, diagnostic_accumulator = nil, test_properties = nil) ⇒ EventProcessor

Returns a new instance of EventProcessor.

Raises:

  • (ArgumentError) 


93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/ldclient-rb/events.rb', line 93

def initialize(sdk_key, config, client = nil, diagnostic_accumulator = nil, test_properties = nil)
  raise ArgumentError, "sdk_key must not be nil" if sdk_key.nil?  # see LDClient constructor comment on sdk_key
  @logger = config.logger
  @inbox = SizedQueue.new(config.capacity < 100 ? 100 : config.capacity)
  @flush_task = Concurrent::TimerTask.new(execution_interval: config.flush_interval) do
    post_to_inbox(FlushMessage.new)
  end
  @flush_task.execute
  @users_flush_task = Concurrent::TimerTask.new(execution_interval: config.user_keys_flush_interval) do
    post_to_inbox(FlushUsersMessage.new)
  end
  @users_flush_task.execute
  if !diagnostic_accumulator.nil?
    interval = test_properties && test_properties.has_key?(:diagnostic_recording_interval) ?
      test_properties[:diagnostic_recording_interval] :
      config.diagnostic_recording_interval
    @diagnostic_event_task = Concurrent::TimerTask.new(execution_interval: interval) do
      post_to_inbox(DiagnosticEventMessage.new)
    end
    @diagnostic_event_task.execute
  else
    @diagnostic_event_task = nil
  end
  @stopped = Concurrent::AtomicBoolean.new(false)
  @inbox_full = Concurrent::AtomicBoolean.new(false)

  event_sender = test_properties && test_properties.has_key?(:event_sender) ?
    test_properties[:event_sender] :
    Impl::EventSender.new(sdk_key, config, client ? client : Util.new_http_client(config.events_uri, config))

  EventDispatcher.new(@inbox, sdk_key, config, diagnostic_accumulator, event_sender)
end

Instance Method Details

#add_event(event) ⇒ Object 



126
127
128
129
 
# File 'lib/ldclient-rb/events.rb', line 126

def add_event(event)
  event[:creationDate] = Impl::Util.current_time_millis
  post_to_inbox(EventMessage.new(event))
end

#flush ⇒ Object 



131
132
133
134
 
# File 'lib/ldclient-rb/events.rb', line 131

def flush
  # flush is done asynchronously
  post_to_inbox(FlushMessage.new)
end

#stop ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/ldclient-rb/events.rb', line 136

def stop
  # final shutdown, which includes a final flush, is done synchronously
  if @stopped.make_true
    @flush_task.shutdown
    @users_flush_task.shutdown
    @diagnostic_event_task.shutdown if !@diagnostic_event_task.nil?
    # Note that here we are not calling post_to_inbox, because we *do* want to wait if the inbox
    # is full; an orderly shutdown can't happen unless these messages are received.
    @inbox << FlushMessage.new
    stop_msg = StopMessage.new
    @inbox << stop_msg
    stop_msg.wait_for_completion
  end
end

#wait_until_inactive ⇒ Object

exposed only for testing



152
153
154
155
156
 
# File 'lib/ldclient-rb/events.rb', line 152

def wait_until_inactive
  sync_msg = TestSyncMessage.new
  @inbox << sync_msg
  sync_msg.wait_for_completion
end