class Backend

Represents a live backend client and its admission state.

Definitions

def initialize(endpoint, client, exchange_limit:, permit_limit: 1, &available)

Initialize a backend.

Signature
parameter　endpoint　Endpoint

The endpoint served by this backend.

parameter　client　Interface(:call, :close)

The HTTP client for the endpoint.

parameter　exchange_limit　Integer

The maximum number of outstanding response exchanges.

parameter　permit_limit　Integer

The maximum number of concurrent processing permits.

yields　{|backend| ...}

Invoked when the backend can accept another request.

Implementation

 
def initialize(endpoint, client, exchange_limit:, permit_limit: 1, &available)
	raise ArgumentError, "Exchange limit must be positive!" unless exchange_limit.positive?
	raise ArgumentError, "Permit limit must be positive!" unless permit_limit.positive?
	
	@endpoint = endpoint
	@client = client
	@exchange_limit = exchange_limit
	@permit_limit = permit_limit
	@available = available
	
	@guard = Thread::Mutex.new
	@active = true
	@processing = 0
	@processing_by_queue = Hash.new(0)
	@exchanges = 0
	@closed = false
end

attr :endpoint

Signature
attribute　Endpoint

The endpoint served by this backend.

def name

Signature
attribute　String

The stable backend name.

Implementation

 
def name
	@endpoint.name
end

attr :exchange_limit

Signature
attribute　Integer

The maximum number of outstanding response exchanges.

def start

Start advertising this backend as available.

Implementation

 
def start
	notify_available
end

def reserve(queue_name = :default)

Reserve the processing slot and one response exchange.

Signature
returns　Boolean

Whether the backend was successfully reserved.

Implementation

 
def reserve(queue_name = :default)
	@guard.synchronize do
		if @active && @processing < @permit_limit && @exchanges < @exchange_limit
			@processing += 1
			@processing_by_queue[queue_name] += 1
			@exchanges += 1
			return true
		end
	end
	
	return false
end

def call(request)

Send a request to the backend.

Signature
parameter　request　Protocol::HTTP::Request

The upstream request.

returns　Protocol::HTTP::Response

The upstream response.

Implementation

 
def call(request)
	@client.call(request)
end

def processed(queue_name = :default)

Release the request-processing slot after response headers arrive.

Implementation

 
def processed(queue_name = :default)
	@guard.synchronize do
		release_processing(queue_name)
	end
	
	notify_available
end

def failed(queue_name = :default)

Release both reservations when a request fails before response headers.

Implementation

 
def failed(queue_name = :default)
	close = @guard.synchronize do
		release_processing(queue_name)
		@exchanges -= 1
		should_close?
	end
	
	notify_available
	close_client if close
end

def release

Release an outstanding response exchange.

Implementation

 
def release
	close = @guard.synchronize do
		raise RuntimeError, "Backend has no outstanding exchange!" unless @exchanges.positive?
		@exchanges -= 1
		should_close?
	end
	
	notify_available
	close_client if close
end

def retire

Retire this backend without interrupting outstanding responses.

Implementation

 
def retire
	close = @guard.synchronize do
		@active = false
		should_close?
	end
	
	close_client if close
end

def active?

Signature
returns　Boolean

Whether this backend still accepts new requests.

Implementation

 
def active?
	@guard.synchronize{@active}
end

def exchanges

Signature
returns　Integer

The number of outstanding response exchanges.

Implementation

 
def exchanges
	@guard.synchronize{@exchanges}
end

def processing?

Signature
returns　Boolean

Whether a request is waiting for response headers.

Implementation

 
def processing?
	@guard.synchronize{@processing.positive?}
end

def processing

Signature
returns　Integer

The number of active processing permits.

Implementation

 
def processing
	@guard.synchronize{@processing}
end

def processing_for(queue_name)

Signature
returns　Integer

The number of active permits for the given queue affinity.

Implementation

 
def processing_for(queue_name)
	@guard.synchronize{@processing_by_queue[queue_name]}
end

def available?

Signature
returns　Boolean

Whether another request can be admitted.

Implementation

 
def available?
	@guard.synchronize{@active && @processing < @permit_limit && @exchanges < @exchange_limit}
end