Class: NatsWork::LoggedConnection

Inherits:
Object
  • Object
show all
Includes:
Logging
Defined in:
lib/natswork/logging.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Logging

included, #log_debug, #log_error, #log_info, #log_warn, #logger

Constructor Details

#initialize(connection) ⇒ LoggedConnection

Returns a new instance of LoggedConnection.



118
119
120
121
 
# File 'lib/natswork/logging.rb', line 118

def initialize(connection)
  @connection = connection
  setup_logging_callbacks
end

Instance Attribute Details

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



116
117
118
 
# File 'lib/natswork/logging.rb', line 116

def connection
  @connection
end

Instance Method Details

#connect ⇒ Object 



123
124
125
126
127
128
129
130
131
 
# File 'lib/natswork/logging.rb', line 123

def connect
  log_info 'Connecting to NATS', servers: @connection.servers
  result = @connection.connect
  log_info 'Connected to NATS successfully'
  result
rescue StandardError => e
  log_error 'Failed to connect to NATS', e, servers: @connection.servers
  raise
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


139
140
141
 
# File 'lib/natswork/logging.rb', line 139

def connected?
  @connection.connected?
end

#disconnect ⇒ Object 



133
134
135
136
137
 
# File 'lib/natswork/logging.rb', line 133

def disconnect
  log_info 'Disconnecting from NATS'
  @connection.disconnect
  log_info 'Disconnected from NATS'
end

#healthy? ⇒ Boolean

Returns:

  • (Boolean) 


206
207
208
 
# File 'lib/natswork/logging.rb', line 206

def healthy?
  @connection.healthy?
end

#jetstream ⇒ Object 



198
199
200
 
# File 'lib/natswork/logging.rb', line 198

def jetstream
  @connection.jetstream
end

#on_disconnect(&block) ⇒ Object 



221
222
223
 
# File 'lib/natswork/logging.rb', line 221

def on_disconnect(&block)
  @connection.on_disconnect(&block)
end

#on_error(&block) ⇒ Object 



225
226
227
 
# File 'lib/natswork/logging.rb', line 225

def on_error(&block)
  @connection.on_error(&block)
end

#on_reconnect(&block) ⇒ Object

Delegate callback methods



217
218
219
 
# File 'lib/natswork/logging.rb', line 217

def on_reconnect(&block)
  @connection.on_reconnect(&block)
end

#ping ⇒ Object 



210
211
212
213
214
 
# File 'lib/natswork/logging.rb', line 210

def ping
  result = @connection.ping
  log_debug 'Ping result', success: result
  result
end

#publish(subject, payload) ⇒ Object 



143
144
145
146
147
148
149
 
# File 'lib/natswork/logging.rb', line 143

def publish(subject, payload)
  log_debug 'Publishing message', subject: subject, payload_size: payload.to_s.bytesize
  @connection.publish(subject, payload)
rescue StandardError => e
  log_error 'Failed to publish message', e, subject: subject
  raise
end

#request(subject, payload, opts = {}) ⇒ Object 



170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/natswork/logging.rb', line 170

def request(subject, payload, opts = {})
  log_debug 'Sending request', subject: subject, timeout: opts[:timeout]
  start_time = Time.now

  response = @connection.request(subject, payload, opts)

  duration = Time.now - start_time
  log_debug 'Received response', subject: subject, duration: duration

  response
rescue NatsWork::TimeoutError
  log_warn 'Request timed out', subject: subject, timeout: opts[:timeout]
  raise
rescue StandardError => e
  log_error 'Request failed', e, subject: subject
  raise
end

#stats ⇒ Object 



202
203
204
 
# File 'lib/natswork/logging.rb', line 202

def stats
  @connection.stats
end

#subscribe(subject, opts = {}, &block) ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/natswork/logging.rb', line 151

def subscribe(subject, opts = {}, &block)
  log_info 'Subscribing to subject', subject: subject, options: opts

  wrapped_block = proc do |msg, reply, subject, sid|
    log_debug 'Received message', subject: subject, reply: reply
    block.call(msg, reply, subject, sid)
  rescue StandardError => e
    log_error 'Error processing message', e, subject: subject
    raise
  end

  sid = @connection.subscribe(subject, opts, &wrapped_block)
  log_info 'Subscribed successfully', subject: subject, sid: sid
  sid
rescue StandardError => e
  log_error 'Failed to subscribe', e, subject: subject
  raise
end

#unsubscribe(sid) ⇒ Object 



188
189
190
191
192
 
# File 'lib/natswork/logging.rb', line 188

def unsubscribe(sid)
  log_debug 'Unsubscribing', sid: sid
  @connection.unsubscribe(sid)
  log_debug 'Unsubscribed successfully', sid: sid
end

#with_connection(&block) ⇒ Object 



194
195
196
 
# File 'lib/natswork/logging.rb', line 194

def with_connection(&block)
  @connection.with_connection(&block)
end