Class: Skyfall::Jetstream::Message Abstract

Inherits:
Object
  • Object
show all
Defined in:
lib/skyfall/jetstream/message.rb

Overview

This class is abstract.

Abstract base class representing a Jetstream message.

Actual messages are returned as instances of one of the subclasses of this class, depending on the type of message, most commonly as CommitMessage.

The Message.new method is overridden here so that it can be called with a JSON message from the websocket, and it parses the type from the JSON and builds an instance of a matching subclass.

You normally don’t need to call this class directly, unless you’re building a custom subclass of Stream or reading raw data packets from the websocket through the Stream#on_raw_message event handler.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(json) ⇒ Message

Returns a new instance of Message.

Parameters:

  • json (Hash) —

    message JSON decoded from the websocket message

Raises:

  • (DecodeError) —

    if the message doesn’t include required data



80
81
82
83
84
85
86
87
 
# File 'lib/skyfall/jetstream/message.rb', line 80

def initialize(json)
  %w(kind did time_us).each { |f| raise DecodeError.new("Missing event details (#{f})") if json[f].nil? }

  @json = json
  @type = @json['kind'].to_sym
  @did = @json['did']
  @time_us = @json['time_us']
end

Instance Attribute Details

#did ⇒ String (readonly) Also known as: repo

DID of the account (repo) that the event is sent by

Returns:

  • (String) 


32
33
34
 
# File 'lib/skyfall/jetstream/message.rb', line 32

def did
  @did
end

#json ⇒ Object (readonly)

The raw JSON of the message as parsed from the websocket packet.



45
46
47
 
# File 'lib/skyfall/jetstream/message.rb', line 45

def json
  @json
end

#time_us ⇒ Integer (readonly) Also known as: seq

Server timestamp of the message (in Unix time microseconds), which serves as a cursor when reconnecting; an equivalent of Firehose::Message#seq in CBOR firehose messages.

Returns:

  • (Integer) 


38
39
40
 
# File 'lib/skyfall/jetstream/message.rb', line 38

def time_us
  @time_us
end

#type ⇒ Symbol (readonly) Also known as: kind

Type of the message (e.g. :commit, :identity etc.)

Returns:

  • (Symbol) 


28
29
30
 
# File 'lib/skyfall/jetstream/message.rb', line 28

def type
  @type
end

Class Method Details

.new(data) ⇒ Skyfall::Jetstream::Message

Parses the JSON data from a websocket message and returns an instance of an appropriate subclass.

UnknownMessage is returned if the message type is not recognized.

Parameters:

  • data (String) —

    plain text payload of a Jetstream websocket message

Returns:

Raises:

  • (DecodeError) —

    if the message doesn’t include required data



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/skyfall/jetstream/message.rb', line 56

def self.new(data)
  json = JSON.parse(data)

  message_class = case json['kind']
    when 'account'  then Jetstream::AccountMessage
    when 'commit'   then Jetstream::CommitMessage
    when 'identity' then Jetstream::IdentityMessage
    else Jetstream::UnknownMessage
  end

  if self != Jetstream::Message && self != message_class
    expected_type = self.name.split('::').last.gsub(/Message$/, '').downcase
    raise DecodeError, "Expected '#{expected_type}' message, got '#{json['kind']}'"
  end

  message = message_class.allocate
  message.send(:initialize, json)
  message
end

Instance Method Details

#operation ⇒ nil Also known as: op

Returns a record operation included in the message. Only :commit messages include operations, but for convenience the method is declared here and returns nil in other messages.

Returns:

  • (nil) 


101
102
103
 
# File 'lib/skyfall/jetstream/message.rb', line 101

def operation
  nil
end

#operations ⇒ Array<Jetstream::Operation>

List of operations on records included in the message. Only :commit messages include operations, but for convenience the method is declared here and returns an empty array in other messages.

Returns:



113
114
115
 
# File 'lib/skyfall/jetstream/message.rb', line 113

def operations
  []
end

#time ⇒ Time

Timestamp decoded from the message.

Note: the time is read from the #time_us field, which stores the event time as an integer in Unix time microseconds, and which is used as an equivalent of Firehose::Message#seq in CBOR firehose messages. This timestamp represents the time when the message was received and stored by Jetstream, which might differ a lot from the created_at time saved in the record data, e.g. if user’s local time is set incorrectly or if an archive of existing posts was imported from another platform. It will also differ (usually only slightly) from the timestamp of the original CBOR message emitted from the PDS and passed through the relay.

Returns:

  • (Time) 


130
131
132
 
# File 'lib/skyfall/jetstream/message.rb', line 130

def time
  @time ||= Time.at(@time_us / 1_000_000.0)
end

#unknown? ⇒ Boolean

Returns true if the message is UnknownMessage (of unrecognized type).

Returns:

  • (Boolean) —

    true if the message is UnknownMessage (of unrecognized type)



92
93
94
 
# File 'lib/skyfall/jetstream/message.rb', line 92

def unknown?
  self.is_a?(Jetstream::UnknownMessage)
end