Module: Protobuf::Nats

Defined in:
lib/protobuf/nats.rb,
lib/protobuf/nats/client.rb,
lib/protobuf/nats/config.rb,
lib/protobuf/nats/errors.rb,
lib/protobuf/nats/runner.rb,
lib/protobuf/nats/server.rb,
lib/protobuf/nats/version.rb,
lib/protobuf/nats/thread_pool.rb,
lib/protobuf/nats/uuidv7_helper.rb,
lib/protobuf/nats/response_muxer.rb,
lib/protobuf/nats/response_muxer_request.rb,
lib/protobuf/nats/super_subscription_manager.rb

Defined Under Namespace

Modules: Errors, Messages Classes: Client, Config, ResponseMuxer, ResponseMuxerRequest, Runner, Server, SuperSubscriptionManager, ThreadPool, UUIDv7Helper

Constant Summary collapse

NatsClient = 
::NATS::IO::Client
GET_CONNECTED_MUTEX = 
::Mutex.new
VERSION = 
"0.13.0.pre0"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.client_nats_connection ⇒ Object

Returns the value of attribute client_nats_connection.



24
25
26
 
# File 'lib/protobuf/nats.rb', line 24

def client_nats_connection
  @client_nats_connection
end

Class Method Details

.config ⇒ Object 



36
37
38
39
40
41
42
 
# File 'lib/protobuf/nats.rb', line 36

def self.config
  @config ||= begin
    config = ::Protobuf::Nats::Config.new
    config.load_from_yml
    config
  end
end

.error_callbacks ⇒ Object

We will always log an error.



48
49
50
 
# File 'lib/protobuf/nats.rb', line 48

def self.error_callbacks
  @error_callbacks ||= [lambda { |error| log_error(error) }]
end

.log_error(error) ⇒ Object 



118
119
120
121
122
123
124
 
# File 'lib/protobuf/nats.rb', line 118

def self.log_error(error)
  logger.error error.to_s
  logger.error error.class.to_s
  if error.respond_to?(:backtrace) && error.backtrace.is_a?(::Array)
    logger.error error.backtrace.join("\n")
  end
end

.logger ⇒ Object 



126
127
128
 
# File 'lib/protobuf/nats.rb', line 126

def self.logger
  ::Protobuf::Logging.logger
end

.notify_error_callbacks(error) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/protobuf/nats.rb', line 61

def self.notify_error_callbacks(error)
  error_callbacks.each do |callback|
    begin
      callback.call(error)
    rescue => callback_error
      log_error(callback_error)
    end
  end

  nil
end

.on_error(&block) ⇒ Object 



55
56
57
58
59
 
# File 'lib/protobuf/nats.rb', line 55

def self.on_error(&block)
  fail ::ArgumentError unless block.arity == 1
  error_callbacks << block
  nil
end

.start_client_nats_connection ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/protobuf/nats.rb', line 81

def self.start_client_nats_connection
  return true if @client_nats_connection

  GET_CONNECTED_MUTEX.synchronize do
    break true if @client_nats_connection

    # Disable publisher pending buffer on reconnect
    options = config.connection_options.merge(:disable_reconnect_buffer => true)

    client = NatsClient.new
    client.connect(options)

    # Ensure we have a valid connection to the NATS server.
    client.flush(5)

    client.on_disconnect do
      logger.warn("Client NATS connection was disconnected")
    end

    client.on_reconnect do
      logger.warn("Client NATS connection was reconnected")
    end

    client.on_close do
      logger.warn("Client NATS connection was closed")
    end

    client.on_error do |error|
      notify_error_callbacks(error)
    end

    @client_nats_connection = client

    true
  end
end

.subscription_key(service_klass, service_method) ⇒ Object 



73
74
75
76
77
78
79
 
# File 'lib/protobuf/nats.rb', line 73

def self.subscription_key(service_klass, service_method)
  service_class_name = service_klass.name.underscore.gsub("/", ".")
  service_method_name = service_method.to_s.underscore

  subscription_key = "rpc.#{service_class_name}.#{service_method_name}"
  subscription_key = config.make_subscription_key_replacements(subscription_key)
end