Class: RExec::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/rexec/connection.rb

Overview

This class represents an abstract connection to another ruby process. The interface does not impose any structure on the way this communication link works, except for the fact you can send and receive objects. You can implement whatever kind of idiom you need for communication on top of this library.

Depending on how you set things up, this could connect to a local ruby process, or a remote ruby process via ‘ssh`.

To set up a connection, you need to use start_server.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input, output, error = nil) ⇒ Connection

Create a new connection. You need to supply a pipe for reading input, a pipe for sending output, and optionally a pipe for errors to be read from.



39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/rexec/connection.rb', line 39

def initialize(input, output, error = nil)
	@input = input
	@output = output

	@running = true
	@exceptions = :send

	@error = error

	@receive_mutex = Mutex.new
	@send_mutex = Mutex.new
end

Instance Attribute Details

#exceptions ⇒ Object

Whether to send exceptions across the wire, or handle normally (e.g. print to stdout):



56
57
58
 
# File 'lib/rexec/connection.rb', line 56

def exceptions
  @exceptions
end

#handler ⇒ Object

The object that will handle remote proxy invocations.



53
54
55
 
# File 'lib/rexec/connection.rb', line 53

def handler
  @handler
end

#proxy ⇒ Object (readonly)

The proxy object that will dispatch RPCs.



59
60
61
 
# File 'lib/rexec/connection.rb', line 59

def proxy
  @proxy
end

Class Method Details

.build(process, options) {|cin| ... } ⇒ Object

Yields:

  • (cin)

Raises:



22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/rexec/connection.rb', line 22

def self.build(process, options, &block)
	cin = process.input
	cout = process.output
	cerr = process.error

	# We require both cin and cout to be connected in order for connection to work
	raise InvalidConnectionError.new("Input (#{cin}) or Output (#{cout}) is not connected!") unless cin and cout

	yield cin

	cin.puts("\004")

	return self.new(cout, cin, cerr)
end

Instance Method Details

#dump_errors(to = $stderr) ⇒ Object

Dump any text which has been written to $stderr in the child process.



117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/rexec/connection.rb', line 117

def dump_errors(to = $stderr)
	if @error and !@error.closed?
		while true
			result = IO.select([@error], [], [], 0)

			break if result == nil

			to.puts @error.readline.chomp
		end
	end
end

#error ⇒ Object

The pipe used for receiving errors. On the client side this pipe is writable, on the server side this pipe is readable. You should avoid using it on the client side and simply use $stderr.



73
74
75
 
# File 'lib/rexec/connection.rb', line 73

def error
	@error
end

#input ⇒ Object

The pipe used for reading data



62
63
64
 
# File 'lib/rexec/connection.rb', line 62

def input
	@input
end

#output ⇒ Object

The pipe used for writing data



67
68
69
 
# File 'lib/rexec/connection.rb', line 67

def output
	@output
end

#receive_object ⇒ Object

Receive an object from the connection. This function is thread-safe. This function may block.



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/rexec/connection.rb', line 130

def receive_object
	object = nil

	@receive_mutex.synchronize do
		begin
			object = Marshal.load(@input)
		rescue EOFError
			object = nil
			@running = false
		end
	end

	if object and object.kind_of?(Exception)
		raise object
	end

	return object
end

#run(&block) ⇒ Object

This is a very simple runloop. It provides an object when it is received.



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/rexec/connection.rb', line 91

def run(&block)
	while @running
		pipes = IO.select([@input])

		if pipes[0].size > 0
			object = receive_object

			if object == nil
				@running = false
				return
			end

			begin
				yield object
			rescue Exception => ex
				if @exceptions == :send
					send_object(ex)
				else
					raise
				end
			end
		end
	end
end

#running? ⇒ Boolean

Return whether or not the connection is running.

Returns:

  • (Boolean) 


86
87
88
 
# File 'lib/rexec/connection.rb', line 86

def running?
	@running
end

#send_object(*objects) ⇒ Object

Send object(s). This function is thread-safe.



150
151
152
153
154
155
156
157
158
159
 
# File 'lib/rexec/connection.rb', line 150

def send_object(*objects)
	@send_mutex.synchronize do
		objects.each do |o|
			data = Marshal.dump(o)
			@output.write(data)
		end

		@output.flush
	end
end

#stop ⇒ Object

Stop the connection, and close the output pipe.



78
79
80
81
82
83
 
# File 'lib/rexec/connection.rb', line 78

def stop
	if @running
		@running = false
		@output.close
	end
end