Module: PubSubTie::Publisher

Extended by:
Publisher
Included in:
Publisher
Defined in:
lib/pubsub_tie/publisher.rb

Instance Method Summary collapse

Instance Method Details

#batch(event_sym, messages, resource) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/pubsub_tie/publisher.rb', line 46

def batch(event_sym, messages, resource)
  topic = @pubsub.
      topic(Events.full_name event_sym)
  messages.each do |data|
    message = augmented(data, event_sym)
    topic.publish_async(payload(validate_data(event_sym, message), resource),
                        publish_time: Time.now.utc) do |result|
      unless result.succeeded?
        Rails.logger.error(
            "Failed to publish #{message} to #{event_sym} on #{resource} due to #{result.error}")
      end
    end
  end
  topic.async_publisher.stop.wait!
end

#configure(config) ⇒ Object 



7
8
9
 
# File 'lib/pubsub_tie/publisher.rb', line 7

def configure(config)
  @pubsub = google_pubsub(config)
end

#google_pubsub(config) ⇒ Object 



11
12
13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/pubsub_tie/publisher.rb', line 11

def google_pubsub(config)
  keyfile_name = config && config['keyfile']
  project_id = (config && config['project_id']) || ENV['PUBSUB_PROJECT'] || ENV['GOOGLE_CLOUD_PROJECT']

  kwargs = {}
  kwargs[:project_id] = project_id if project_id

  if keyfile_name && !keyfile_name.to_s.strip.empty?
    keyfile = File.expand_path(keyfile_name, File.join(PubSubTie.app_root, 'config'))
    kwargs[:credentials] = ::Google::Cloud::PubSub::Credentials.new(keyfile)
  end

  ::Google::Cloud::PubSub.new(**kwargs)
end

#publish(event_sym, data, resource) ⇒ Object

Publishes event data asynchronously to topic inferred from event_sym. Data is augmented with event_name and event_time and validated against loaded configuration



31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/pubsub_tie/publisher.rb', line 31

def publish(event_sym, data, resource)
  message = augmented(data, event_sym)      

  @pubsub.
    topic(Events.full_name event_sym).
    # publish(message(payload, resource), publish_time: Time.now.utc)
    publish_async(payload(validate_data(event_sym, message), resource),
                  publish_time: Time.now.utc) do |result|
      unless result.succeeded?
        Rails.logger.error(
          "Failed to publish #{message} to #{event_sym} on #{resource} due to #{result.error}")
      end
    end
end