Class: Takagi::Controller

Inherits:
Object
  • Object
show all
Extended by:
Base::ReactorManagement
Defined in:
lib/takagi/controller.rb,
lib/takagi/controller/thread_pool.rb,
lib/takagi/controller/resource_allocator.rb,
sig/takagi/controller.rbs

Overview

Base class for modular controllers with isolated routers

Controllers provide route isolation, independent process pools, and nested mounting capabilities for building scalable CoAP applications.

Examples:

Simple controller

class TelemetryController < Takagi::Controller
  configure do
    mount '/telemetry'
    profile :high_throughput
  end

  post '/data' do
    # POST /telemetry/data
  end
end

Nested controllers

class ApiController < Takagi::Controller
  configure do
    mount '/api'
    nest DevicesController, UsersController
  end
end

class DevicesController < Takagi::Controller
  configure do
    mount '/devices'
    profile :high_throughput
  end
end

Direct Known Subclasses

Application::DiscoveryController

Defined Under Namespace

Classes: ConfigContext, ResourceAllocator, ThreadPool

Class Method Summary collapse

Methods included from Base::ReactorManagement

reactor, start_reactors, stop_reactors, use_reactor

Class Method Details

.add_route(method, path, metadata: {}) {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Register a route

Parameters:

  • method (String)
  • path (String)
  • metadata: (Hash[Symbol, untyped]) (defaults to: {})

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


88
89
90
 
# File 'lib/takagi/controller.rb', line 88

def add_route(method, path, metadata: {}, &block)
  router.add_route(method, path, metadata: metadata, &block)
end

.config ⇒ config_hash

Get the controller's configuration

Returns:

  • (config_hash) 


57
58
59
60
61
62
63
64
65
66
 
# File 'lib/takagi/controller.rb', line 57

def config
  @config ||= {
    mount_path: nil,
    nested_from: nil,
    nested_controllers: [],
    profile: nil,
    processes: nil,
    threads: nil
  }
end

.configure {|arg0| ... } ⇒ void

This method returns an undefined value.

Configure the controller

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


78
79
80
 
# File 'lib/takagi/controller.rb', line 78

def configure(&block)
  ConfigContext.new(self).instance_eval(&block) if block
end

.delete {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


30
 
# File 'sig/takagi/controller.rbs', line 30

def self.delete: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.fetch {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


32
 
# File 'sig/takagi/controller.rbs', line 32

def self.fetch: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.get {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

HTTP/CoAP method shortcuts (get, post, put, delete, etc.)

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


27
 
# File 'sig/takagi/controller.rbs', line 27

def self.get: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.ipatch {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


33
 
# File 'sig/takagi/controller.rbs', line 33

def self.ipatch: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.mount_path ⇒ String?

Get the full mount path

Returns:

  • (String, nil) 


118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/takagi/controller.rb', line 118

def mount_path
  path = config[:mount_path]
  return nil unless path

  # If nested from parent, prepend parent's mount path
  if config[:nested_from]
    parent_path = config[:nested_from].mount_path
    return File.join(parent_path, path) if parent_path
  end

  path
end

.mounted? ⇒ Boolean

Check if mounted

Returns:

  • (Boolean) 


134
135
136
 
# File 'lib/takagi/controller.rb', line 134

def mounted?
  !config[:mount_path].nil?
end

.nested_controllers ⇒ Array[singleton(Controller)]

Get nested controllers

Returns:



141
142
143
 
# File 'lib/takagi/controller.rb', line 141

def nested_controllers
  config[:nested_controllers]
end

.observable(path, metadata: {}) {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Observable route registration

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped]) (defaults to: {})

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


108
109
110
111
 
# File 'lib/takagi/controller.rb', line 108

def observable(path, metadata: {}, &block)
  observable_metadata = { obs: true, rt: 'core#observable', if: 'takagi.observe' }
  add_route('OBSERVE', path, metadata: observable_metadata.merge(metadata), &block)
end

.patch {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


31
 
# File 'sig/takagi/controller.rbs', line 31

def self.patch: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.post {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


28
 
# File 'sig/takagi/controller.rbs', line 28

def self.post: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.process_count ⇒ Integer?

Get effective process count

Returns:

  • (Integer, nil) 


157
158
159
 
# File 'lib/takagi/controller.rb', line 157

def process_count
  config[:processes] || profile_config[:processes]
end

.profile_name ⇒ Symbol?

Get profile name

Returns:

  • (Symbol, nil) 


148
149
150
 
# File 'lib/takagi/controller.rb', line 148

def profile_name
  config[:profile]
end

.put {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • path (String)
  • metadata: (Hash[Symbol, untyped])

Yields:

Yield Parameters:

Yield Returns:

  • (Object) 


29
 
# File 'sig/takagi/controller.rbs', line 29

def self.put: (String path, ?metadata: Hash[Symbol, untyped]) { (Message::Inbound, Hash[Symbol, String]) -> untyped } -> void

.router ⇒ Router

Get the controller's isolated router

Returns:



50
51
52
 
# File 'lib/takagi/controller.rb', line 50

def router
  @router ||= Router.new
end

.schedule { ... } ⇒ Object

Schedule a job to run in the controller's thread pool

Yields:

  • Block to execute in worker thread

Raises:

  • (ThreadPoolError) —

    if thread pool not started



212
213
214
215
216
217
218
219
 
# File 'lib/takagi/controller.rb', line 212

def schedule(&block)
  unless @thread_pool
    error = Errors::ThreadPoolError.not_started(name)
    raise error
  end

  @thread_pool.schedule(&block)
end

.shutdown_workers! ⇒ Object

Shutdown the controller's worker thread pool



199
200
201
202
203
204
205
206
 
# File 'lib/takagi/controller.rb', line 199

def shutdown_workers!
  return unless @thread_pool

  Takagi.logger.info "Shutting down worker pool for #{name}"
  Takagi::Hooks.emit(:controller_workers_stopped, controller: self)
  @thread_pool.shutdown
  @thread_pool = nil
end

.start_workers!(threads: nil, name: nil) ⇒ ThreadPool

Start the controller's worker thread pool

Parameters:

  • threads (Integer) (defaults to: nil) —

    Number of threads to allocate

  • name (String) (defaults to: nil) —

    Pool name (defaults to controller class name)

Returns:



188
189
190
191
192
193
194
195
196
 
# File 'lib/takagi/controller.rb', line 188

def start_workers!(threads: nil, name: nil)
  threads ||= thread_count || 4  # Default to 4 threads
  name ||= self.name.split('::').last  # e.g., "IngressController"

  @thread_pool = ThreadPool.new(size: threads, name: name)
  Takagi.logger.info "Started worker pool for #{name} with #{threads} threads"
  Takagi::Hooks.emit(:controller_workers_started, controller: self, name: name, threads: threads)
  @thread_pool
end

.thread_count ⇒ Integer?

Get effective thread count

Returns:

  • (Integer, nil) 


166
167
168
 
# File 'lib/takagi/controller.rb', line 166

def thread_count
  config[:threads] || profile_config[:threads]
end

.thread_pool ⇒ ThreadPool

Get or create the controller's thread pool

Lazy initialization: creates pool on first access if not already started. This allows reactors to share the controller's pool automatically.

Returns:



176
177
178
179
180
181
 
# File 'lib/takagi/controller.rb', line 176

def thread_pool
  @thread_pool ||= begin
    Takagi.logger.debug "Lazy-initializing thread pool for #{name}"
    start_workers!
  end
end

.workers_running? ⇒ Boolean

Check if the controller's thread pool is running

Returns:

  • (Boolean) —

    true if thread pool is active



224
225
226
 
# File 'lib/takagi/controller.rb', line 224

def workers_running?
  @thread_pool && !@thread_pool.shutdown?
end