Avoid torn writes on notify pipe.

Author: ioquatix

lib/async/container/channel.rb

@@ -10,8 +10,9 @@ module Container
 		# Provides a basic multi-thread/multi-process uni-directional communication channel.
 		class Channel
 			# Initialize the channel using a pipe.
-			def initialize
+			def initialize(timeout: 1.0)
 				@in, @out = ::IO.pipe
+				@in.timeout = timeout
 			end
 
 			# The input end of the pipe.
@@ -43,12 +44,11 @@ def close
 			# @returns [Hash]
 			def receive
 				if data = @in.gets
-					begin
-						return JSON.parse(data, symbolize_names: true)
-					rescue
-						return {line: data}
-					end
+					return JSON.parse(data, symbolize_names: true)
 				end
+			rescue => error
+				Console.error(self, "Error during channel receive!", error)
+				return nil
 			end
 		end
 	end

lib/async/container/forked.rb

@@ -136,8 +136,8 @@ def self.spawn(*arguments, name: nil, **options)
 
 				# Initialize the process.
 				# @parameter name [String] The name to use for the child process.
-				def initialize(name: nil)
-					super()
+				def initialize(name: nil, **options)
+					super(**options)
 
 					@name = name
 					@status = nil

lib/async/container/notify/pipe.rb

@@ -63,9 +63,9 @@ def before_spawn(arguments, options)
 				# Formats the message using JSON and sends it to the parent controller.
 				# This is suitable for use with {Channel}.
 				def send(**message)
-					data = ::JSON.dump(message)
+					data = ::JSON.dump(message) << "\n"
 
-					@io.puts(data)
+					@io.write(data)
 					@io.flush
 				end
 

lib/async/container/threaded.rb

@@ -124,8 +124,8 @@ def self.fork(**options)
 				# Initialize the thread.
 				#
 				# @parameter name [String] The name to use for the child thread.
-				def initialize(name: nil)
-					super()
+				def initialize(name: nil, **options)
+					super(**options)
 
 					@status = nil
 

test/async/container/channel.rb

@@ -29,4 +29,12 @@
 
 		expect(channel.receive).to be == {line: "Hello, World!\n"}
 	end
+	
+	with "timeout" do
+		let(:channel) {subject.new(timeout: 0.001)}
+		
+		it "fails gracefully on timeout" do
+			expect(channel.receive).to be_nil
+		end
+	end
 end

test/async/container/notify/pipe.rb

@@ -30,4 +30,14 @@
 
 		expect(container.statistics).to have_attributes(failures: be == 0)
 	end
+	
+	it "writes data with one write call" do
+		output = StringIO.new
+		expect(output).to receive(:write).with("{\"ready\":true,\"status\":\"All systems go!\"}\n")
+			
+		client = Async::Container::Notify::Pipe.new(output)
+		client.send(ready: true, status: "All systems go!")
+		
+		expect(output.string).to be == "{\"ready\":true,\"status\":\"All systems go!\"}\n"
+	end
 end