Class: Featureflip::DataSource::StreamingHandler

Inherits:
Object
  • Object
show all
Defined in:
lib/featureflip/data_source/streaming.rb

Defined Under Namespace

Classes: StreamStopped

Constant Summary collapse

SERVER_PING_INTERVAL_SECONDS =

The server sends a keep-alive ping this often; a finite read timeout at or below this interval would sever a healthy stream.

30
STREAM_READ_TIMEOUT =

Client-side liveness watchdog. net/http gives us no separate heartbeat, so the read timeout IS the watchdog: if no data (not even a ping) arrives for this long the socket is treated as dead — a half-open connection (LB/NAT idle-drop or a partition with no FIN/RST) — and the blocking read raises Net::ReadTimeout, which drives reconnect/backoff/polling. Set to 3× the ping (3 missed pings) so it never severs a healthy stream but still detects a dead socket within a bounded time. MUST stay finite and

SERVER_PING_INTERVAL_SECONDS. (The rest of the family runs an infinite read timeout — java readTimeout(0) / python read=None / csharp #1526 — and has the same latent half-open hole; ruby closes it here.)

SERVER_PING_INTERVAL_SECONDS * 3
RECONNECT_BASE_DELAY_SECONDS =

Base reconnect backoff; also the floor applied after a healthy stream closes cleanly, so even an accept-then-immediately-close server is throttled instead of busy-looping.

1
MAX_BACKOFF_SECONDS = 
30

Instance Method Summary collapse

Constructor Details

#initialize(sdk_key:, config:, http_client:, on_flag_updated:, on_flag_deleted:, on_segment_updated:, on_error:, on_sync: nil, on_give_up: nil) ⇒ StreamingHandler

Returns a new instance of StreamingHandler.



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/featureflip/data_source/streaming.rb', line 39

def initialize(sdk_key:, config:, http_client:, on_flag_updated:, on_flag_deleted:, on_segment_updated:, on_error:, on_sync: nil, on_give_up: nil)
  @sdk_key = sdk_key
  @config = config
  @http_client = http_client
  @on_flag_updated = on_flag_updated
  @on_flag_deleted = on_flag_deleted
  @on_segment_updated = on_segment_updated
  @on_error = on_error
  @on_sync = on_sync
  @on_give_up = on_give_up
  @stop_flag = false
  @thread = nil
  @retry_count = 0
  @current_event_type = nil
  @current_data = nil
  @line_buffer = String.new # ASCII-8BIT: raw read_body bytes concatenate safely
  @delivered_frame = false
  @wake_mutex = Mutex.new
  @wake_cond = ConditionVariable.new
end

Instance Method Details

#start ⇒ Object 



60
61
62
63
64
 
# File 'lib/featureflip/data_source/streaming.rb', line 60

def start
  @stop_flag = false
  @retry_count = 0
  @thread = Thread.new { run }
end

#stop ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/featureflip/data_source/streaming.rb', line 66

def stop
  @stop_flag = true
  # Wake an in-progress backoff wait.
  @wake_mutex.synchronize { @wake_cond.broadcast }

  thread = @thread
  @thread = nil
  return unless thread

  # Interrupt a thread blocked in read_body. Guard the raise: the thread may
  # finish between the alive? check and the raise (ThreadError on a dead one).
  begin
    thread.raise(StreamStopped.new) if thread.alive?
  rescue ThreadError
    # Thread already finished — nothing to interrupt.
  end
  thread.join(5)
end