Class: Cod::Pipe

Inherits:
Channel show all
Defined in:
lib/cod/pipe.rb

Overview

A cod channel based on IO.pipe.

NOTE: If you embed Cod::Pipe channels into your messages, Cod will insert the object id of that channel into the byte stream that is transmitted. On receiving such an object id (a machine pointer), Cod will try to reconstruct the channel that was at the origin of the id. This can obviously only work if you have such an object in your address space. There are multiple ways to construct such a situation. Say you want to send a pipe channel to one of your (forked) childs: This will work if you create the channel before forking the child, since master and child will share all objects that were available before the fork.

Defined Under Namespace

Modules: SplitMethods

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Channel

#interact

Constructor Details

#initialize(serializer = nil, pipe_pair = nil) ⇒ Pipe

Returns a new instance of Pipe.



28
29
30
31
32
 
# File 'lib/cod/pipe.rb', line 28

def initialize(serializer=nil, pipe_pair=nil)
  super()
  @serializer = serializer || SimpleSerializer.new
  @pipe = IOPair.new(*pipe_pair)
end

Instance Attribute Details

#pipe ⇒ Object (readonly)

Returns the value of attribute pipe.



16
17
18
 
# File 'lib/cod/pipe.rb', line 16

def pipe
  @pipe
end

#serializer ⇒ Object (readonly)

Returns the value of attribute serializer.



17
18
19
 
# File 'lib/cod/pipe.rb', line 17

def serializer
  @serializer
end

Class Method Details

._load(string) ⇒ Object

:nodoc:



167
168
169
 
# File 'lib/cod/pipe.rb', line 167

def self._load(string) # :nodoc:
  ObjectSpace._id2ref(Integer(string))
end

Instance Method Details

#_dump(depth) ⇒ Object

———————————————————- serialization



164
165
166
 
# File 'lib/cod/pipe.rb', line 164

def _dump(depth) # :nodoc:
  object_id.to_s
end

#can_read? ⇒ Boolean

Returns true if you can read from this pipe.

Returns:

  • (Boolean) 


130
131
132
 
# File 'lib/cod/pipe.rb', line 130

def can_read?
  not r.nil?
end

#can_write? ⇒ Boolean

Returns true if you can write to this pipe.

Returns:

  • (Boolean) 


136
137
138
 
# File 'lib/cod/pipe.rb', line 136

def can_write?
  not pipe.w.nil?
end

#client(answers_to) ⇒ Object 



159
160
161
 
# File 'lib/cod/pipe.rb', line 159

def client(answers_to)
  Service::Client.new(self, answers_to)
end

#close ⇒ Object

Closes the pipe completely. All active ends are closed. Note that you can call this function on a closed pipe without getting an error raised.



114
115
116
 
# File 'lib/cod/pipe.rb', line 114

def close
  pipe.close
end

#get(opts = {}) ⇒ Object

Using #get on a pipe instance will close the other pipe end. Subsequent #put will receive a Cod::InvalidOperation.

Example: 

pipe.get # => obj



98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/cod/pipe.rb', line 98

def get(opts={})
  raise Cod::WriteOnlyChannel unless can_read?
  pipe.close_w
  
  loop do
    ready = Cod.select(nil, self)
    return deserialize_one if ready
  end
rescue EOFError
  fail "All pipe ends seem to be closed. Reading from this pipe will not "+
    "return any data."
end

#initialize_copy(other) ⇒ Object

Creates a copy of this pipe channel. This performs a shallow #dup except for the file descriptors stored in the pipe, so that a #close affects only one copy.



38
39
40
41
42
 
# File 'lib/cod/pipe.rb', line 38

def initialize_copy(other)
  super
  @serializer = other.serializer
  @pipe = other.pipe.dup
end

#put(obj) ⇒ Object

Using #put on a pipe instance will close the other pipe end. Subsequent #get will raise a Cod::InvalidOperation.

Example: 

pipe.put [:a, :message]

Raises:



85
86
87
88
89
90
 
# File 'lib/cod/pipe.rb', line 85

def put(obj)
  raise Cod::ReadOnlyChannel unless can_write?
  
  pipe.write(
    serializer.en(obj))
end

#r ⇒ Object

Returns the read end of the pipe



144
145
146
 
# File 'lib/cod/pipe.rb', line 144

def r
  pipe.r
end

#readonly ⇒ Object

Makes this pipe readonly. Calls to #put will error out. This closes the write end permanently and provokes end of file on the read end once all processes that posses a link to the write end do so.

Returns self so that you can write for example: 

read_end = pipe.dup.readonly



51
52
53
54
 
# File 'lib/cod/pipe.rb', line 51

def readonly
  pipe.close_w
  self
end

#select(timeout = nil) ⇒ Object

Returns if this pipe is ready for reading.



120
121
122
123
 
# File 'lib/cod/pipe.rb', line 120

def select(timeout=nil)
  result = Cod.select(timeout, self)
  not result.nil?
end

#service ⇒ Object

——————————————————— service/client



156
157
158
 
# File 'lib/cod/pipe.rb', line 156

def service
  Service.new(self)
end

#split ⇒ Object

Actively splits this pipe into two ends, a read end and a write end. The original pipe is closed, leaving only the two ends to work with. The read end can only be read from (#get) and the write end can only be written to (#put).



71
72
73
74
75
76
77
 
# File 'lib/cod/pipe.rb', line 71

def split
  [self.dup.readonly, self.dup.writeonly].tap { |split|
    self.close
    
    split.extend(SplitMethods)
  }
end

#to_read_fds ⇒ Object 



124
125
126
 
# File 'lib/cod/pipe.rb', line 124

def to_read_fds
  r
end

#w ⇒ Object

Returns the write end of the pipe



150
151
152
 
# File 'lib/cod/pipe.rb', line 150

def w
  pipe.w
end

#writeonly ⇒ Object

Makes this pipe writeonly. Calls to #get will error out. See #readonly.

Returns self so that you can write for example: 

write_end = pipe.dup.writeonly



61
62
63
64
 
# File 'lib/cod/pipe.rb', line 61

def writeonly
  pipe.close_r
  self
end