Module: Rubyists::Leopard::NatsApiServer::WorkerLifecycle

Defined in:
lib/leopard/nats_api_server.rb

Overview

Instance-side worker boot and shutdown helpers.

Instance Method Summary collapse

Instance Method Details

#add_endpoints(endpoints, group_map) ⇒ void (private)

This method returns an undefined value.

Adds endpoints to the NATS service.

Parameters:

  • The list of endpoints to add.

  • A map of group names to their created group objects.

Raises:

  • If an endpoint references an undefined group.



428
429
430
431
432
433
434
435
436
 
# File 'lib/leopard/nats_api_server.rb', line 428

def add_endpoints(endpoints, group_map)
  endpoints.each do |ep|
    grp = ep.group
    parent = grp ? group_map[grp] : @service
    raise ArgumentError, "Group #{grp} not defined" if grp && parent.nil?

    build_endpoint(parent, ep)
  end
end

#add_groups(gps) ⇒ Hash (private)

Adds groups to the NATS service.

Parameters:

  • The groups to add, where keys are group names and values are group definitions.

Returns:

  • A map of group names to their created group objects.



397
398
399
400
401
 
# File 'lib/leopard/nats_api_server.rb', line 397

def add_groups(gps)
  created = {}
  gps.each_key { |name| build_group(gps, created, name) }
  created
end

#build_group(defs, cache, name) ⇒ NATS::Group (private)

Builds a group in the NATS service.

Parameters:

  • The group definitions, where keys are group names and values are group definitions.

  • A cache to store already created groups.

  • The name of the group to build.

Returns:

  • The created group object.

Raises:

  • If the requested group was never defined.



411
412
413
414
415
416
417
418
419
 
# File 'lib/leopard/nats_api_server.rb', line 411

def build_group(defs, cache, name)
  return cache[name] if cache.key?(name)

  gdef = defs[name]
  raise ArgumentError, "Group #{name} not defined" unless gdef

  parent = gdef[:parent] ? build_group(defs, cache, gdef[:parent]) : @service
  cache[name] = parent.groups.add(gdef[:name], queue: gdef[:queue])
end

#build_service_opts(service_opts:) ⇒ Hash (private)

Builds the service options for the NATS service.

Parameters:

  • Options for the NATS service.

Returns:

  • The complete service options including name and version.



385
386
387
388
389
390
 
# File 'lib/leopard/nats_api_server.rb', line 385

def build_service_opts(service_opts:)
  {
    name: self.class.name.split('::').join('.'),
    version: '0.1.0',
  }.merge(service_opts)
end

#connect_client(nats_url) ⇒ Object (private)

Opens the NATS client connection for this worker.

Parameters:

  • The URL of the NATS server.

Returns:

  • The connected NATS client.



313
314
315
 
# File 'lib/leopard/nats_api_server.rb', line 313

def connect_client(nats_url)
  @client = NATS.connect(nats_url)
end

#initialize_service(service_opts) ⇒ Object (private)

Registers the NATS service for this worker.

Parameters:

  • Options for the NATS service.

Returns:

  • The created NATS service.



322
323
324
 
# File 'lib/leopard/nats_api_server.rb', line 322

def initialize_service(service_opts)
  @service = @client.services.add(build_service_opts(service_opts:))
end

#initialize_worker_state ⇒ Thread (private)

Captures the current thread for later wakeup during shutdown.

Returns:

  • The current worker thread.



304
305
306
 
# File 'lib/leopard/nats_api_server.rb', line 304

def initialize_worker_state
  @thread = Thread.current
end

#jetstream_consumer_class ⇒ Class (private)

Returns the JetStream consumer coordinator class for this worker.

Returns:

  • The JetStream consumer implementation class.



369
370
371
 
# File 'lib/leopard/nats_api_server.rb', line 369

def jetstream_consumer_class
  NatsJetstreamConsumer
end

#logger ⇒ Object

Returns the logger configured for the NATS API server.

Returns:

  • The configured logger.



258
 
# File 'lib/leopard/nats_api_server.rb', line 258

def logger = self.class.logger

#setup_worker(nats_url: 'nats://localhost:4222', service_opts: {}) ⇒ void

This method returns an undefined value.

Sets up a worker thread for the NATS API server. This method connects to the NATS server, adds the service, groups, and endpoints,

Parameters:

  • (defaults to: 'nats://localhost:4222')

    The URL of the NATS server.

  • (defaults to: {})

    Options for the NATS service.



267
268
269
270
271
272
273
 
# File 'lib/leopard/nats_api_server.rb', line 267

def setup_worker(nats_url: 'nats://localhost:4222', service_opts: {})
  initialize_worker_state
  connect_client(nats_url)
  initialize_service(service_opts)
  add_endpoints(self.class.endpoints.dup, add_groups(self.class.groups.dup))
  start_jetstream_consumer(self.class.jetstream_endpoints.dup)
end

#setup_worker!(nats_url: 'nats://localhost:4222', service_opts: {}) ⇒ void

This method returns an undefined value.

Sets up a worker thread for the NATS API server and blocks the current thread.

Parameters:

  • (defaults to: 'nats://localhost:4222')

    The URL of the NATS server.

  • (defaults to: {})

    Options for the NATS service.

See Also:



282
283
284
285
 
# File 'lib/leopard/nats_api_server.rb', line 282

def setup_worker!(nats_url: 'nats://localhost:4222', service_opts: {})
  setup_worker(nats_url:, service_opts:)
  sleep
end

#start_jetstream_consumer(endpoints) ⇒ void (private)

This method returns an undefined value.

Starts the JetStream consumer coordinator when JetStream endpoints are present.

Parameters:

  • JetStream endpoints for this worker.



331
332
333
334
335
336
337
338
339
340
341
342
 
# File 'lib/leopard/nats_api_server.rb', line 331

def start_jetstream_consumer(endpoints)
  return if endpoints.empty?

  @jetstream_consumer = jetstream_consumer_class.new(
    jetstream: @client.jetstream,
    endpoints:,
    logger:,
    process_message: method(:process_transport_message),
    thread_factory:,
  )
  @jetstream_consumer.start
end

#stop ⇒ void

This method returns an undefined value.

Stops the NATS API server worker.



290
291
292
293
294
295
296
297
 
# File 'lib/leopard/nats_api_server.rb', line 290

def stop
  @running = false
  stop_jetstream
  stop_service
  wake_worker
rescue ThreadError
  nil
end

#stop_jetstream ⇒ void (private)

This method returns an undefined value.

Stops the JetStream consumer coordinator if one was started.



347
348
349
 
# File 'lib/leopard/nats_api_server.rb', line 347

def stop_jetstream
  @jetstream_consumer&.stop
end

#stop_service ⇒ void (private)

This method returns an undefined value.

Stops the registered NATS service and closes the client connection.



354
355
356
357
 
# File 'lib/leopard/nats_api_server.rb', line 354

def stop_service
  @service&.stop
  @client&.close
end

#thread_factory ⇒ Class (private)

Returns the thread factory used for JetStream consumer loops.

Returns:

  • The thread factory class.



376
377
378
 
# File 'lib/leopard/nats_api_server.rb', line 376

def thread_factory
  Thread
end

#wake_worker ⇒ Thread? (private)

Wakes the worker thread if it is blocked.

Returns:

  • The awakened worker thread, if present.



362
363
364
 
# File 'lib/leopard/nats_api_server.rb', line 362

def wake_worker
  @thread&.wakeup
end