mackuba.eu
A Ruby gem for streaming data from the Bluesky/ATProto firehose
removed faye_ext temporary hack
-48
-48
lib/skyfall/faye_ext.rb
-48
lib/skyfall/faye_ext.rb
···
1
-
require 'websocket/driver'
2
-
require_relative 'firehose'
3
-
4
-
module WebSocket
5
-
class Driver
6
-
class Hybi
7
-
def emit_message
8
-
message = @extensions.process_incoming_message(@message)
9
-
@message = nil
10
-
11
-
payload = message.data
12
-
13
-
case message.opcode
14
-
when OPCODES[:text] then
15
-
payload = Driver.encode(payload, Encoding::UTF_8)
16
-
payload = nil unless payload.valid_encoding?
17
-
# when OPCODES[:binary]
18
-
# payload = payload.bytes.to_a
19
-
end
20
-
21
-
if payload
22
-
emit(:message, MessageEvent.new(payload))
23
-
else
24
-
fail(:encoding_error, 'Could not decode a text frame as UTF-8')
25
-
end
26
-
rescue ::WebSocket::Extensions::ExtensionError => error
27
-
fail(:extension_error, error.message)
28
-
end
29
-
end
30
-
end
31
-
end
32
-
33
-
module Skyfall
34
-
class Firehose
35
-
def handle_message(msg)
36
-
data = msg.data #.pack('C*')
37
-
@handlers[:raw_message]&.call(data)
38
-
39
-
if @handlers[:message]
40
-
atp_message = Message.new(data)
41
-
@cursor = atp_message.seq
42
-
@handlers[:message].call(atp_message)
43
-
else
44
-
@cursor = nil
45
-
end
46
-
end
47
-
end
48
-
end