Class: Protobuf::Nats::Server
- Inherits:
-
Object
- Object
- Protobuf::Nats::Server
- Includes:
- Logging, Rpc::Server
- Defined in:
- lib/protobuf/nats/server.rb
Constant Summary collapse
- MILLISECOND =
1000
Instance Attribute Summary collapse
-
#nats ⇒ Object
readonly
Returns the value of attribute nats.
-
#subscriptions ⇒ Object
readonly
Returns the value of attribute subscriptions.
-
#thread_pool ⇒ Object
readonly
Returns the value of attribute thread_pool.
Instance Method Summary collapse
- #detect_and_handle_a_pause ⇒ Object
- #do_not_subscribe_to_includes?(subscription_key) ⇒ Boolean
- #enqueue_request(request_data, reply_id) ⇒ Object
-
#finish_slow_start ⇒ Object
Slow start subscriptions by adding X rounds of subz every Y seconds, where X is subscriptions_per_rpc_endpoint and Y is slow_start_delay.
-
#initialize(options) ⇒ Server
constructor
A new instance of Server.
- #max_queue_size ⇒ Object
- #only_subscribe_to_includes?(subscription_key) ⇒ Boolean
- #pause_file_path ⇒ Object
- #paused? ⇒ Boolean
- #print_subscription_keys ⇒ Object
- #run ⇒ Object
- #running? ⇒ Boolean
- #service_klasses ⇒ Object
- #slow_start_delay ⇒ Object
- #stop ⇒ Object
- #subscribe ⇒ Object
- #subscribe_to_services_once ⇒ Object
- #subscriptions_per_rpc_endpoint ⇒ Object
- #unsubscribe ⇒ Object
- #with_each_subscription_key ⇒ Object
Constructor Details
#initialize(options) ⇒ Server
Returns a new instance of Server.
17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
# File 'lib/protobuf/nats/server.rb', line 17 def initialize() @options = @processing_requests = true @running = true @stopped = false @nats = @options[:client] || ::Protobuf::Nats::NatsClient.new @nats.connect(::Protobuf::Nats.config.) @thread_pool = ::Protobuf::Nats::ThreadPool.new(@options[:threads], :max_queue => max_queue_size) @subscriptions = [] @server = .fetch(:server, ::Socket.gethostname) end |
Instance Attribute Details
#nats ⇒ Object (readonly)
Returns the value of attribute nats.
13 14 15 |
# File 'lib/protobuf/nats/server.rb', line 13 def nats @nats end |
#subscriptions ⇒ Object (readonly)
Returns the value of attribute subscriptions.
13 14 15 |
# File 'lib/protobuf/nats/server.rb', line 13 def subscriptions @subscriptions end |
#thread_pool ⇒ Object (readonly)
Returns the value of attribute thread_pool.
13 14 15 |
# File 'lib/protobuf/nats/server.rb', line 13 def thread_pool @thread_pool end |
Instance Method Details
#detect_and_handle_a_pause ⇒ Object
161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 |
# File 'lib/protobuf/nats/server.rb', line 161 def detect_and_handle_a_pause case # If we are taking requests and detect a pause file, then unsubscribe. when @processing_requests && paused? @processing_requests = false logger.warn("Pausing server!") unsubscribe # If we were paused and the pause file is no longer present, then subscribe again. when !@processing_requests && !paused? logger.warn("Resuming server: resubscribing to all services and restarting slow start!") @processing_requests = true subscribe end end |
#do_not_subscribe_to_includes?(subscription_key) ⇒ Boolean
85 86 87 88 89 90 91 92 |
# File 'lib/protobuf/nats/server.rb', line 85 def do_not_subscribe_to_includes?(subscription_key) return false unless ::Protobuf::Nats.config.server_subscription_key_do_not_subscribe_to_when_includes_any_of.respond_to?(:any?) return false if ::Protobuf::Nats.config.server_subscription_key_do_not_subscribe_to_when_includes_any_of.empty? ::Protobuf::Nats.config.server_subscription_key_do_not_subscribe_to_when_includes_any_of.any? do |key| subscription_key.include?(key) end end |
#enqueue_request(request_data, reply_id) ⇒ Object
48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
# File 'lib/protobuf/nats/server.rb', line 48 def enqueue_request(request_data, reply_id) ::ActiveSupport::Notifications.instrument "server.message_received.protobuf-nats" enqueued_at = ::Time.now was_enqueued = thread_pool.push do begin # Instrument the thread pool time-to-execute duration. processed_at = ::Time.now ::ActiveSupport::Notifications.instrument("server.thread_pool_execution_delay.protobuf-nats", (processed_at - enqueued_at) * MILLISECOND) # Process request. response_data = handle_request(request_data, 'server' => @server) # Publish response. nats.publish(reply_id, response_data) rescue => error ::Protobuf::Nats.notify_error_callbacks(error) ensure # Instrument the request duration. completed_at = ::Time.now ::ActiveSupport::Notifications.instrument("server.request_duration.protobuf-nats", (completed_at - enqueued_at) * MILLISECOND) end end # Publish an ACK to signal the server has picked up the work. if was_enqueued nats.publish(reply_id, ::Protobuf::Nats::Messages::ACK) else ::ActiveSupport::Notifications.instrument "server.message_dropped.protobuf-nats" nats.publish(reply_id, ::Protobuf::Nats::Messages::NACK) end was_enqueued end |
#finish_slow_start ⇒ Object
Slow start subscriptions by adding X rounds of subz every Y seconds, where X is subscriptions_per_rpc_endpoint and Y is slow_start_delay.
144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
# File 'lib/protobuf/nats/server.rb', line 144 def finish_slow_start logger.info "Slow start has started..." completed = 1 # We have (X - 1) here because we always subscribe at least once. (subscriptions_per_rpc_endpoint - 1).times do next unless @running next if paused? completed += 1 sleep slow_start_delay subscribe_to_services_once logger.info "Slow start adding another round of subscriptions (#{completed}/#{subscriptions_per_rpc_endpoint})..." end logger.info "Slow start finished." end |
#max_queue_size ⇒ Object
32 33 34 |
# File 'lib/protobuf/nats/server.rb', line 32 def max_queue_size ::ENV.fetch("PB_NATS_SERVER_MAX_QUEUE_SIZE", @options[:threads]).to_i end |
#only_subscribe_to_includes?(subscription_key) ⇒ Boolean
94 95 96 97 98 99 100 101 |
# File 'lib/protobuf/nats/server.rb', line 94 def only_subscribe_to_includes?(subscription_key) return true unless ::Protobuf::Nats.config.server_subscription_key_only_subscribe_to_when_includes_any_of.respond_to?(:any?) return true if ::Protobuf::Nats.config.server_subscription_key_only_subscribe_to_when_includes_any_of.empty? ::Protobuf::Nats.config.server_subscription_key_only_subscribe_to_when_includes_any_of.any? do |key| subscription_key.include?(key) end end |
#pause_file_path ⇒ Object
103 104 105 |
# File 'lib/protobuf/nats/server.rb', line 103 def pause_file_path ::ENV.fetch("PB_NATS_SERVER_PAUSE_FILE_PATH", nil) end |
#paused? ⇒ Boolean
177 178 179 |
# File 'lib/protobuf/nats/server.rb', line 177 def paused? !pause_file_path.nil? && ::File.exist?(pause_file_path) end |
#print_subscription_keys ⇒ Object
107 108 109 110 111 112 113 |
# File 'lib/protobuf/nats/server.rb', line 107 def print_subscription_keys logger.info "Creating subscriptions:" with_each_subscription_key do |subscription_key| logger.info " - #{subscription_key}" end end |
#run ⇒ Object
181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
# File 'lib/protobuf/nats/server.rb', line 181 def run nats.on_reconnect do logger.warn "Server NATS connection was reconnected" end nats.on_disconnect do logger.warn "Server NATS connection was disconnected" end nats.on_error do |error| ::Protobuf::Nats.notify_error_callbacks(error) end nats.on_close do logger.warn "Server NATS connection was closed" end print_subscription_keys if paused? yield if block_given? else subscribe { yield if block_given? } end loop do break unless @running detect_and_handle_a_pause sleep 1 end unsubscribe logger.info "Waiting up to 60 seconds for the thread pool to finish shutting down..." thread_pool.shutdown thread_pool.wait_for_termination(60) ensure @stopped = true end |
#running? ⇒ Boolean
220 221 222 |
# File 'lib/protobuf/nats/server.rb', line 220 def running? @stopped end |
#service_klasses ⇒ Object
44 45 46 |
# File 'lib/protobuf/nats/server.rb', line 44 def service_klasses ::Protobuf::Rpc::Service.implemented_services.map(&:safe_constantize) end |
#slow_start_delay ⇒ Object
36 37 38 |
# File 'lib/protobuf/nats/server.rb', line 36 def slow_start_delay @slow_start_delay ||= ::ENV.fetch("PB_NATS_SERVER_SLOW_START_DELAY", 10).to_i end |
#stop ⇒ Object
224 225 226 |
# File 'lib/protobuf/nats/server.rb', line 224 def stop @running = false end |
#subscribe ⇒ Object
228 229 230 231 232 |
# File 'lib/protobuf/nats/server.rb', line 228 def subscribe subscribe_to_services_once yield if block_given? finish_slow_start end |
#subscribe_to_services_once ⇒ Object
115 116 117 118 119 120 121 122 123 |
# File 'lib/protobuf/nats/server.rb', line 115 def subscribe_to_services_once with_each_subscription_key do |subscription_key_and_queue| subscriptions << nats.subscribe(subscription_key_and_queue, :queue => subscription_key_and_queue) do |request_data, reply_id, _subject| unless enqueue_request(request_data, reply_id) logger.error { "Thread pool is full! Dropping message for: #{subscription_key_and_queue}" } end end end end |
#subscriptions_per_rpc_endpoint ⇒ Object
40 41 42 |
# File 'lib/protobuf/nats/server.rb', line 40 def subscriptions_per_rpc_endpoint @subscriptions_per_rpc_endpoint ||= ::ENV.fetch("PB_NATS_SERVER_SUBSCRIPTIONS_PER_RPC_ENDPOINT", 10).to_i end |
#unsubscribe ⇒ Object
234 235 236 237 238 239 |
# File 'lib/protobuf/nats/server.rb', line 234 def unsubscribe logger.info "Unsubscribing from rpc routes..." subscriptions.each do |subscription_id| nats.unsubscribe(subscription_id) end end |
#with_each_subscription_key ⇒ Object
125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 |
# File 'lib/protobuf/nats/server.rb', line 125 def with_each_subscription_key fail ::ArgumentError unless block_given? service_klasses.each do |service_klass| service_klass.rpcs.each do |service_method, _| # Skip services that are not implemented. next unless service_klass.method_defined?(service_method) subscription_key = ::Protobuf::Nats.subscription_key(service_klass, service_method) next if do_not_subscribe_to_includes?(subscription_key) next unless only_subscribe_to_includes?(subscription_key) yield subscription_key end end end |