Class: LanGrove::Plugin::Enqueuer

Inherits:
Base
  • Object
show all
Defined in:
lib/langrove/plugin/enqueuer.rb

Instance Attribute Summary

Attributes inherited from Base

#logger

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Base

#config_exception, #type

Constructor Details

#initialize(root, config, name = nil) ⇒ Enqueuer

Returns a new instance of Enqueuer.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/langrove/plugin/enqueuer.rb', line 15

def initialize( root, config, name = nil )
  
  super( root, config, name )
  
  validate_config
  
  @provides = {
    
    :enqueuer => {
      
      :version => ENQUEUER_VERSION
      
    }
    
  }

end

Class Method Details

.type ⇒ Object 



9
10
11
12
13
 
# File 'lib/langrove/plugin/enqueuer.rb', line 9

def self.type
  
  :enqueuer
  
end

Instance Method Details

#enqueue_capsule(handler) ⇒ Object

Raises:



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/langrove/plugin/enqueuer.rb', line 53

def enqueue_capsule( handler )
  
  #
  # Enqueue a job into a work queue
  # Insert the capsule.capsule as payload
  # 
  
  header = queue_header( :load_capsule, handler )
  data = handler.capsule
  
  # enqueue( header, data )

  raise LanGrove::PluginException.new(

    "#{self.class}.enqueue_casule() not implemented."
    
  )

end

#enqueue_capsule_handler(handler) ⇒ Object 



83
84
85
86
87
88
89
90
91
 
# File 'lib/langrove/plugin/enqueuer.rb', line 83

def enqueue_capsule_handler( handler )
  
  @root.trigger( handler, :handler, :enqueue )
  
  enqueue_capsule( handler )
  
  @root.trigger( handler, :handler, :after_enqueue )
  
end

#enqueue_key(handler) ⇒ Object

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/langrove/plugin/enqueuer.rb', line 33

def enqueue_key( handler )
  
  #
  # Enqueue a job into a work queue
  # Insert the capsule.key as payload
  # 
  
  header = queue_header( :get_capsule, handler )
  data = handler.key
  
  # enqueue( header, data )

  raise LanGrove::PluginException.new(

    "#{self.class}.enqueue_key() not implemented."
    
  )

end

#enqueue_key_handler(handler) ⇒ Object 



73
74
75
76
77
78
79
80
81
 
# File 'lib/langrove/plugin/enqueuer.rb', line 73

def enqueue_key_handler( handler )
  
  @root.trigger( handler, :handler, :enqueue )
  
  enqueue_key( handler )
  
  @root.trigger( handler, :handler, :after_enqueue )
  
end

#queue_header(action, handler, routing_flags = 'none') ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/langrove/plugin/enqueuer.rb', line 93

def queue_header( action, handler, routing_flags = 'none' )
  
  #
  # Jobs inserted into a work queue should contain 
  # as first paremeter the following header:
  # 
  #

  context = "#{@root.config.system}:#{@root.config.application}:#{routing_flags}"
  
  return {

    'context' => context,
    
    #
    # Required - A timestamp, 
    # 
    # - Allows for monitoring of queue lag.
    # 
    # - Allows for a future QueueMarshal which
    #   can redirect by age into a :do_immediately
    #   and a :do_later_because_old_data_isnot_urgent
    #   
    #
    'created_at'  => Time.now,
    
    #
    # Optional - A dequeue action,
    # 
    # The worker dequeueing the Job will make a call 
    # to whatever function is specified ahead of
    # yielding the Job to be processed
    #
    #
    'action' => action,
    
    #
    # Optional - A sequence number, 
    # 
    # - Allows for the ability to 'expire' jobs.
    # 
    # eg.  A sequence on the persisted capsule differing 
    #      from the sequence in the key on the associated 
    #      Job as queued would mean a subsequent update to 
    #      the persisted capsule has ocurred, 
    #      
    #      And an additional Job has been queued to process 
    #      it, so the first job can therefore be skipped in 
    #      the cases where:
    # 
    #      A. Only the most recent state is relevant 
    #         for processing.
    #  
    #      B. The capsule is persisted from a single 
    #         source, all other processes access it
    #         from a read only capacity.
    #
    #      Z. The Job is quick, or sufficiently not slow,
    #         or being queued by something possibly faster, 
    #         but also queueing jobs against enough separate
    #         persistables in a manner that will ensure that 
    #         the possibility of the queue worker never ever
    #         catching up to a 'latest' capsule, and therefore
    #         permanently collecting and skipping jobs, is
    #         sufficiently unlikely.
    # 
    #
    'sequence' => handler.capsule['persisted_seq']
    
  }
  
end

#validate_config ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/langrove/plugin/enqueuer.rb', line 166

def validate_config
  
  #
  # Deferrer requires :worker specified
  # as the WorkerClass that will consume
  # jobs from the queue
  #
  
  config_exception( 
  
    "#{self.class} requires :job:" 
    
  ) unless @job = @config[ :job ]
  
end