Ruby gem for ingesting ATProto repo data from a Tap service (extension of Skyfall gem)
sending acks
+22
-3
+3
lib/tapfall/errors.rb
+3
lib/tapfall/errors.rb
+19
-3
lib/tapfall/stream.rb
+19
-3
lib/tapfall/stream.rb
···
3
3
require_relative 'messages/tap_message'
4
4
5
5
class Tapfall::Stream < Skyfall::Stream
6
-
def initialize(server)
6
+
def initialize(server, options = {})
7
7
super(server)
8
8
9
+
@options = options
9
10
@root_url = ensure_empty_path(@root_url)
11
+
@ack = true unless options[:ack] == false
10
12
end
11
13
12
-
def handle_message(msg)
13
-
data = msg.data
14
+
def connect
15
+
if @ack && @handlers[:message].nil?
16
+
raise ConfigError, "The on(:message) handler must be set unless :ack => false option is passed"
17
+
end
18
+
19
+
super
20
+
end
21
+
22
+
def handle_message(packet)
23
+
data = packet.data
14
24
@handlers[:raw_message]&.call(data)
15
25
16
26
if @handlers[:message]
17
27
tap_message = Tapfall::TapMessage.new(data)
18
28
@handlers[:message].call(tap_message)
29
+
send_ack(tap_message) if @ack
19
30
end
31
+
end
32
+
33
+
def send_ack(msg)
34
+
json = JSON.generate(type: 'ack', id: msg.id)
35
+
send_data(json)
20
36
end
21
37
22
38
private