Class: Polyn::PullSubscriber

Inherits:
Object
  • Object
show all
Defined in:
lib/polyn/pull_subscriber.rb

Overview

Wrapper around nats-pure that can validate polyn messages

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(fields) ⇒ PullSubscriber

is more than the ‘source_root`

Parameters:

  • fields (Object) —

    :nats - Connected NATS instance from ‘NATS.connect`

  • fields (String) —

    :type - The type of event

Options Hash (fields):

  • :source (String) —

    • If the ‘source` portion of the consumer name



12
13
14
15
16
17
18
19
20
 
# File 'lib/polyn/pull_subscriber.rb', line 12

def initialize(fields)
  @nats          = fields.fetch(:nats)
  @type          = Polyn::Naming.trim_domain_prefix(fields.fetch(:type))
  @consumer_name = Polyn::Naming.consumer_name(@type, fields[:source])
  @stream        = @nats.jetstream.find_stream_name_by_subject(@type)
  self.class.validate_consumer_exists!(@nats, @stream, @consumer_name)
  @psub          = @nats.jetstream.pull_subscribe(@type, @consumer_name)
  @serializer    = fields.fetch(:serializer)
end

Class Method Details

.validate_consumer_exists!(nats, stream, consumer_name) ⇒ Object

nats-pure will create a consumer if the one you passed does not exist. Polyn wants to avoid this functionality and instead encourage consumer creation in the centralized ‘events` codebase so that it’s documented, discoverable, and polyn-cli can manage it



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

def self.validate_consumer_exists!(nats, stream, consumer_name)
  nats.jetstream.consumer_info(stream, consumer_name)
rescue NATS::JetStream::Error::NotFound
  raise Polyn::Errors::ValidationError,
    "Consumer #{consumer_name} does not exist. Use polyn-cli to create"\
    "it before attempting to subscribe"
end

Instance Method Details

#fetch(batch = 1, params = {}) ⇒ Array<NATS::Msg>

fetch makes a request to be delivered more messages from a pull consumer.

Parameters:

  • batch (Fixnum) (defaults to: 1) —

    Number of messages to pull from the stream.

  • params (Hash) (defaults to: {}) —

    Options to customize the fetch request.

Options Hash (params):

  • :timeout (Float) —

    Duration of the fetch request before it expires.

Returns:

  • (Array<NATS::Msg>) 


40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/polyn/pull_subscriber.rb', line 40

def fetch(batch = 1, params = {})
  Polyn::Tracing.processing_span(@type) do |process_span|
    msgs = @psub.fetch(batch, params)
    msgs.map do |msg|
      Polyn::Tracing.subscribe_span(@type, msg, links: [process_span]) do |span|
        updated_msg = process_message(msg)
        Polyn::Tracing.span_attributes(span,
          nats:    @nats,
          type:    @type,
          event:   updated_msg.data,
          payload: msg.data)
        updated_msg
      end
    end
  end
end