Module: Plushie::Runtime::Subscriptions

Included in:
Plushie::Runtime
Defined in:
lib/plushie/runtime/subscriptions.rb,
sig/plushie/runtime.rbs

Overview

Subscription lifecycle management for the Plushie runtime.

Compares the app's subscribe(model) output against active subscriptions, starting new ones and stopping removed ones. Timer subscriptions run locally; renderer subscriptions are forwarded to the bridge.

Instance Method Summary collapse

Instance Method Details

#check_max_rate(key, spec)

This method returns an undefined value.

Check if a single subscription's max_rate needs updating.

Parameters:



137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/plushie/runtime/subscriptions.rb', line 137

def check_max_rate(key, spec)
  entry = @subscriptions[key]
  return unless entry && entry[:sub_type] == :renderer && entry[:max_rate] != spec.max_rate

  # Re-send subscribe with new rate
  wire_tag = spec.wire_tag
  @bridge.send_encoded(
    Protocol::Encode.encode_subscribe(spec.type, wire_tag, @format,
      max_rate: spec.max_rate, window_id: spec.window_id)
  )
  @subscriptions[key] = entry.merge(max_rate: spec.max_rate, wire_tag: wire_tag)
end

#diff_subscriptions(new_by_key, new_sorted_keys)

This method returns an undefined value.

Full diff of subscription sets.

Parameters:



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/plushie/runtime/subscriptions.rb', line 53

def diff_subscriptions(new_by_key, new_sorted_keys)
  old_keys = @subscriptions.keys.to_set
  new_keys = new_by_key.keys.to_set

  # Stop removed subscriptions
  (old_keys - new_keys).each { |key| stop_subscription(key) }

  # Start new subscriptions
  new_entries = {}
  (new_keys - old_keys).each do |key|
    spec = new_by_key[key]
    new_entries[key] = start_subscription(spec)
  end

  # Keep existing (check max_rate changes)
  kept = {}
  (new_keys & old_keys).each do |key|
    kept[key] = @subscriptions[key]
    check_max_rate(key, new_by_key[key])
  end

  @subscriptions = kept.merge(new_entries)
  @subscription_keys = new_sorted_keys
end

#start_renderer_subscription(spec) ⇒ Hash[Symbol, untyped]

Start a renderer subscription (send subscribe message to bridge).

Parameters:

Returns:

  • (Hash[Symbol, untyped]) 


117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/plushie/runtime/subscriptions.rb', line 117

def start_renderer_subscription(spec)
  wire_tag = spec.wire_tag

  @bridge.send_encoded(
    Protocol::Encode.encode_subscribe(spec.type, wire_tag, @format,
      max_rate: spec.max_rate, window_id: spec.window_id)
  )

  {sub_type: :renderer, kind: spec.type, wire_tag: wire_tag,
   max_rate: spec.max_rate, window_id: spec.window_id}
end

#start_subscription(spec) ⇒ Hash[Symbol, untyped]

Start a new subscription (timer or renderer).

Parameters:

Returns:

  • (Hash[Symbol, untyped]) 


79
80
81
82
83
84
85
 
# File 'lib/plushie/runtime/subscriptions.rb', line 79

def start_subscription(spec)
  if spec.type == :every
    start_timer_subscription(spec)
  else
    start_renderer_subscription(spec)
  end
end

#start_timer_subscription(spec) ⇒ Hash[Symbol, untyped]

Start a timer subscription (runs locally, pushes to event queue).

Parameters:

Returns:

  • (Hash[Symbol, untyped]) 


107
108
109
110
111
112
113
114
 
# File 'lib/plushie/runtime/subscriptions.rb', line 107

def start_timer_subscription(spec)
  tag = spec.tag
  interval = spec.interval

  @timer_scheduler.schedule(tag: tag, interval_ms: interval, event_queue: @event_queue)

  {sub_type: :timer, tag: tag, interval: interval}
end

#stop_subscription(key)

This method returns an undefined value.

Stop a subscription by key.

Parameters:

  • key (Object) 


88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/plushie/runtime/subscriptions.rb', line 88

def stop_subscription(key)
  entry = @subscriptions.delete(key)
  return unless entry

  case entry[:sub_type]
  when :timer
    @timer_scheduler.cancel(entry[:tag])
  when :renderer
    @bridge.send_encoded(
      Protocol::Encode.encode_unsubscribe(
        entry[:kind],
        tag: entry[:wire_tag],
        format: @format
      )
    )
  end
end

#sync_subscriptions

This method returns an undefined value.

Synchronize subscriptions with the app's current subscribe output. Called after each update cycle.



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/plushie/runtime/subscriptions.rb', line 17

def sync_subscriptions
  new_specs = begin
    subs = @app.subscribe(@model)
    subs = subs.is_a?(Array) ? subs : []
    subs.select do |s|
      if s.is_a?(Subscription::Sub)
        true
      else
        @logger.warn("plushie: subscribe returned invalid spec (dropping): #{s.inspect}")
        false
      end
    end
  rescue => e
    @logger.error("plushie: subscribe raised: #{e.class}: #{e.message}")
    []
  end

  # Merge canvas widget subscriptions
  cw = defined?(@canvas_widgets) ? @canvas_widgets : nil
  if cw && !cw.empty?
    widget_subs = CanvasWidget.collect_subscriptions(cw)
    new_specs += widget_subs
  end

  new_by_key = new_specs.each_with_object({}) { |spec, h| h[spec.key] = spec }
  new_sorted_keys = new_by_key.keys.sort_by(&:to_s)

  if new_sorted_keys == @subscription_keys
    # Short-circuit: key set unchanged, just check max_rate updates
    update_max_rates(new_by_key)
  else
    diff_subscriptions(new_by_key, new_sorted_keys)
  end
end

#update_max_rates(new_by_key)

This method returns an undefined value.

Update max_rate on existing renderer subscriptions if changed.

Parameters:



130
131
132
133
134
 
# File 'lib/plushie/runtime/subscriptions.rb', line 130

def update_max_rates(new_by_key)
  new_by_key.each do |key, spec|
    check_max_rate(key, spec)
  end
end