Class: Cosmo::Publisher

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/cosmo/publisher.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Publisher

Returns a new instance of Publisher.



17
18
19
 
# File 'lib/cosmo/publisher.rb', line 17

def initialize
  @client = Client.instance
end

Class Method Details

.instance ⇒ Object 



13
14
15
 
# File 'lib/cosmo/publisher.rb', line 13

def self.instance
  @instance ||= new
end

Instance Method Details

#publish(subject, data, serializer: nil, **options) ⇒ Object 



21
22
23
24
 
# File 'lib/cosmo/publisher.rb', line 21

def publish(subject, data, serializer: nil, **options)
  payload = (serializer || Stream::Serializer).serialize(data)
  @client.publish(subject, payload, **options)
end

#publish_batch(subject, batch, **options) ⇒ Object 



34
35
36
 
# File 'lib/cosmo/publisher.rb', line 34

def publish_batch(subject, batch, **options)
  batch.each { publish(subject, _1, **options) }
end

#publish_job(data) ⇒ Object 



26
27
28
29
30
31
32
 
# File 'lib/cosmo/publisher.rb', line 26

def publish_job(data)
  subject, payload, params = data.to_args
  @client.publish(subject, payload, **params)
  data.jid
rescue NATS::JetStream::Error::NoStreamResponse
  raise StreamNotFoundError, params[:stream].to_s
end