/usr/lib/ruby/vendor_ruby/packet/packet_meta_pimp.rb is in ruby-packet 0.1.15-5.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 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 | # Class acts as a pimp for workers, which doesn't have a manually created pimp
# The idea behind a manually added pimp is to let client handle low level messaging
# beween workers. A meta pimp, does it for you.
class Packet::MetaPimp < Packet::Pimp
# initializer of pimp
attr_accessor :callback_hash
attr_accessor :worker_status, :worker_key,:worker_name
attr_accessor :invokable_worker_methods
def pimp_init
@callback_hash ||= {}
@worker_status = nil
@worker_result = nil
@worker_key = nil
@tokenizer = Packet::BinParser.new
end
# will be invoked whenever there is a response from the worker
def receive_data p_data
@tokenizer.extract(p_data) do |b_data|
t_data = Marshal.load(b_data)
handle_object(t_data)
end
end
def handle_object data_options = {}
case data_options[:type]
when :request
process_request(data_options)
when :response
process_response(data_options)
when :status
save_worker_status(data_options)
when :result
save_worker_result(data_options)
end
end
def save_worker_result(data_options = { })
@worker_result = data_options[:data]
end
def save_worker_status(data_options = { })
# @worker_status = data_options[:data]
reactor.update_result(worker_key,data_options[:data])
end
def process_request(data_options = {})
if((requested_worker = data_options[:requested_worker]) && (reactor.live_workers[requested_worker]))
reactor.live_workers[requested_worker].send_request(data_options)
end
end
def process_response(data_options = {})
if callback_signature = data_options[:callback_signature]
callback = callback_hash[callback_signature]
# there coule be bug when you are trying to send the data back to the client
begin
callback.invoke(data_options)
rescue
end
elsif client_signature = data_options[:client_signature]
begin
reactor.connections[client_signature][:instance].worker_receive(data_options)
rescue
end
end
end
# can be used to send request to correspoding worker
def send_request(data_options = { })
if callback = data_options[:callback]
callback_hash[callback.signature] = callback
data_options.delete(:callback)
data_options[:callback_signature] = callback.signature
send_data(data_options)
else
send_data(data_options)
end
end
end
|