Class: PRuby::Pipeline Abstract

Inherits:
Object show all
Defined in:
lib/pruby/pipeline.rb

Overview

This class is abstract.

Un pipeline est compose d'une suite de noeuds (aka. "nodes", "stages", etages) connectes entre eux. Un pipeline possede un canal d'entree et un canal de sortie. On peut lancer l'execution d'un pipeline (avec run) puis attendre qu'il se termine (avec join). On peut aussi connaitre la/les valeur/s finale/s retournee/s par son/ses threads associes (avec value).

Un Pipeline est en fait une classe purement abstraite, puisque chaque noeud d'un Pipeline est aussi une instance d'une sous-classe de Pipeline. Il s'agit donc d'un cas typique du patron "composite", illustre par le diagramme de classes ci-joint.

                   ___________________
                   |                 |  1..*
                   |    Pipeline     |/___________________________
                   |_________________|\                           |
                            /\                                    |
                           ----                                   |
                             |                                    |
                             |                                    |
        |--------------------|---------------------|              |
        |                    |                     |              |
________|__________  ________|__________   ________|__________    |
|                 |  |                 |   |                 |    |
|    ProcNode     |  |  PipelineNode   |   |   FarmNode      |    |
|_________________|  |_________________|   |_________________|    |
                              |                    |              |
                              |--------------------|---------------

Direct Known Subclasses

FarmNode, PipelineNode, ProcNode

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#input_channel ⇒ Channel Also known as: input

This method is abstract.

Le canal d'entree

Returns:



75
76
77
 
# File 'lib/pruby/pipeline.rb', line 75

def input_channel
  fail "*** La methode input_channel doit etre redefinie dans une sous-classe"
end

#output_channel ⇒ Channel Also known as: output

This method is abstract.

Le canal de sortie

Returns:



84
85
86
 
# File 'lib/pruby/pipeline.rb', line 84

def output_channel
  fail "*** La methode output_channel doit etre redefinie dans une sous-classe"
end

Class Method Details

.create(*args) ⇒ Pipeline

Cree un pipeline. Simple operation de facade.

Parameters:

  • args (liste de Pipeline ou Proc) —

    Liste des noeuds a inclure dans le nouveau pipeline

Returns:

  • (Pipeline) —

    Le pipeline nouvellement cree



55
56
57
 
# File 'lib/pruby/pipeline.rb', line 55

def self.create( *args )
  PipelineFactory.pipeline *args
end

.sink(sink) ⇒ Pipeline

Cree un puits pour un pipeline. Simple operation de facade.

Parameters:

  • sink (Array, String) —

    Le puits, i.e., la destination des elements

Returns:

  • (Pipeline) —

    Un noeud pouvant etre utilise (seulement) comme dernier etage d’un pipeline

Ensures:

  • si sink.class == Array alors les elements recus sur le canal d’entree seront ajoutes au tableau sink

  • si sink.class == String alors les elements recus sur le canal d’entree seront ajoutes au fichier dont le nom est sink

  • self.sink?



66
67
68
 
# File 'lib/pruby/pipeline.rb', line 66

def self.sink( sink )
  PipelineFactory.sink sink
end

.source(source, source_kind = nil) ⇒ Pipeline

Cree une source pour un pipeline. Simple operation de facade.

Parameters:

  • source (#each, String) —

    La source qui fournira les elements a emettre sur le canal de sortie

  • source_kind (nil, :string, :filename) (defaults to: nil) —

    La sorte de source

Returns:

  • (Pipeline) —

    Un noeud pouvant etre utilise (seulement) comme premier etage d’un pipeline

Ensures:

  • self.source?



44
45
46
 
# File 'lib/pruby/pipeline.rb', line 44

def self.source( source, source_kind = nil )
  PipelineFactory.source source, source_kind
end

Instance Method Details

#>>(other) ⇒ self

Note:

Identique a |. Utilise simplement parce que plus “image”

Note:

Attention: la priorite de “>>” differe de celle de “|” !?!?

Connecte une source a un pipeline ou un pipeline a un puits.

Returns:

  • (self)

Requires:

  • L’un ou l’autre des deux elements a connecter doit etre une source ou un puits



147
148
149
150
151
 
# File 'lib/pruby/pipeline.rb', line 147

def >>( other )
  DBC.require( source? || other.class == Proc || other.sink?,
               "*** Cette operation ne peut etre utilisee qu'avec une source (a gauche) ou un sink (a droite)" )
  add_stage( other )
end

#input_channel_connected? ⇒ Bool

Determine si le canal d'entree du pipeline est connecte.

Returns:

  • (Bool) 


105
106
107
 
# File 'lib/pruby/pipeline.rb', line 105

def input_channel_connected?
  input_channel
end

#join ⇒ self

Bloque jusqu'a ce que l'execution du/des noeud/s soit terminee. Si deja termine, alors NOOP.

Returns:

  • (self)

Ensures:

  • L’execution des noeuds internes est terminee



203
204
205
206
207
208
209
210
 
# File 'lib/pruby/pipeline.rb', line 203

def join
  return if terminated?

  inner_nodes.map(&:join)
  @terminated = true

  self
end

#output_channel_connected? ⇒ Bool

Determine si le canal de sortie du pipeline est connecte.

Returns:

  • (Bool) 


112
113
114
 
# File 'lib/pruby/pipeline.rb', line 112

def output_channel_connected?
  output_channel
end

#run(no_wait = nil) ⇒ self

Lance un/des thread/s pour executer la/les tache/s associee/s au/x noeud/s. Donc, s'il s'agit d'un noeud composite, alors lance l'execution des noeuds internes.

Parameters:

  • no_wait (Bool) (defaults to: nil) —

    Si :NO_WAIT, alors on n’attend pas que l’execution des noeuds internes soit terminee pour retourner

Returns:

  • (self)

Ensures:

  • Si no_wait != :NO_WAIT, alors l’execution des noeuds internes est terminee



188
189
190
191
192
193
194
195
 
# File 'lib/pruby/pipeline.rb', line 188

def run( no_wait = nil )
  @terminated = false

  inner_nodes.map { |s| s.run :NO_WAIT }
  join unless no_wait == :NO_WAIT

  self
end

#sink? ⇒ Bool

Determine si le noeud est un puits, donc qui n'emet rien sur le canal de sortie.

Returns:

  • (Bool) 


126
127
128
 
# File 'lib/pruby/pipeline.rb', line 126

def sink?
  @source_ou_sink == :sink
end

#source? ⇒ Bool

Determine si le noeud est une source, donc qui ne recoit rien du canal d'entree.

Returns:

  • (Bool) 


119
120
121
 
# File 'lib/pruby/pipeline.rb', line 119

def source?
  @source_ou_sink == :source
end

#terminated? ⇒ Bool

Indique si le thread associe au pipeline a termine son execution

Returns:

  • (Bool) 


156
157
158
 
# File 'lib/pruby/pipeline.rb', line 156

def terminated?
  @terminated
end

#wrap_around! ⇒ self

Cree un lien de feedback, avec un nouveau canal, entre la sortie et l'entree du pipeline

Returns:

  • (self)

Ensures:

  • Un nouveau canal connecte la sortie a l’entree

Requires:

  • Les canaux d’entre et de sortie ne doivent pas deja etre connectes



168
169
170
171
172
173
174
175
176
 
# File 'lib/pruby/pipeline.rb', line 168

def wrap_around!
  DBC.require !input_channel_connected?,  "*** L'entree du pipeline est deja connectee"
  DBC.require !output_channel_connected?, "*** La sortie du pipeline est deja connectee"

  chan = Channel.new
  self.output_channel = chan
  self.input_channel = chan
  self
end

#|(other) ⇒ self

Connecte deux pipelines entre eux.

Parameters:

  • other (Pipeline) —

    L’autre pipeline a connecter avec le premier (self)

Returns:

  • (self)

Ensures:

  • Ajoute other comme etage (noeud) additionel au pipeline courant



136
137
138
 
# File 'lib/pruby/pipeline.rb', line 136

def |( other )
  add_stage( other )
end