/[storm]/storm/doc/pegboard/storm_with_tapestry--hemppah/peg.rst
ViewVC logotype

Diff of /storm/doc/pegboard/storm_with_tapestry--hemppah/peg.rst

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.17 by hemppah, Tue Jul 8 08:24:38 2003 UTC revision 1.18 by hemppah, Tue Jul 8 09:17:36 2003 UTC
# Line 157  network: Line 157  network:
157  An event handler class  An event handler class
158  ------------------  ------------------
159    
160  1. Write an eventhandler class which extends SEDA's existing stages, e.g.,  1. Write an event handler class which extends SEDA's existing stages, e.g.,
161    
162    public class StormTapestryManager implements EventHandlerIF    public class StormTapestryManager implements EventHandlerIF
163        
164  2. In a eventhandler class, implement a hanleEvent method which will  2. In an event handler class, implement a hanleEvent method which will
165     process registered messages     process registered messages
166        
167  3. In a eventhandler class, implement a init method which will define  3. In an eventhandler class, implement a init method which will define
168     the events and messages that are "listened" for this class     the events and messages that are "listened" for this class
169    
170  4. Register your custom message classes  4. Register your custom message classes
# Line 172  An event handler class Line 172  An event handler class
172  Using Tapestry with Storm  Using Tapestry with Storm
173  =========================  =========================
174    
175  In this section, we will outline the design which would allow Storm to  In this section, we will outline a design which would allow Storm to use
176  Tapestry's routing services.  Tapestry's routing services.
177    
178  For finding Storm blocks in a Tapestry overlay, we must create custom message  For finding Storm blocks in a Tapestry overlay, we must create custom message
179  classes:  classes:
180    
181     - BlockIDTag class for Storm blocks' IDs (implements     - BlockIDTag class (implements ostore.tapestry.api.TapestryTag) for
182       ostore.tapestry.api.TapestryTag) for storing query specific information:       Storm blocks' IDs for storing query specific information:
183       in our case, we store at least block's GUID and other related info       in our case, we store at least block's GUID and other related info
184       (if needed)       (if needed)
185    
186     - BlockIDQuery (implements ostore.tapestry.api.TapestryQuery) to receive an     - BlockIDQuery class (implements ostore.tapestry.api.TapestryQuery) to receive an
187       object identified by a BlockIDTag, dispatch a       object identified by a BlockIDTag (and dispatch a
188       ostore.tapestry.api.TapestryLocateMsg that contains a BlockIDQuery         ostore.tapestry.api.TapestryLocateMsg that contains a BlockIDQuery)  
189            
190     - StormLocateMsg class (extends ostore.tapestry.api.TapestryLocateMsg,     - StormLocateMsg class (extends ostore.tapestry.api.TapestryLocateMsg,
191       implements ostore.util.QuickSerializable) for locating Storm blocks       implements ostore.util.QuickSerializable) for locating Storm blocks
# Line 194  classes: Line 194  classes:
194     - StormResponseMsg class (extends ostore.tapestry.api.TapestryRouteMsg,     - StormResponseMsg class (extends ostore.tapestry.api.TapestryRouteMsg,
195       implements ostore.util.QuickSerializable) to carry meta data for a       implements ostore.util.QuickSerializable) to carry meta data for a
196       requested Storm block. This is an optional class, i.e., if we want       requested Storm block. This is an optional class, i.e., if we want
197       to see query results, we may use this       to see query results, we could use this class
198            
199     - StormBlockRequestMsg (extends ostore.tapestry.api.TapestryRouteMsg,     - StormBlockRequestMsg class (extends ostore.tapestry.api.TapestryRouteMsg,
200       implements ostore.util.QuickSerializable) to request a specific Storm       implements ostore.util.QuickSerializable) to request a specific Storm
201       block       block
202        
203     - StormErrorMsg (extends ostore.tapestry.api.TapestryRouteMsg,     - StormErrorMsg class (extends ostore.tapestry.api.TapestryRouteMsg,
204       implements ostore.util.QuickSerializable) for sending error messages       implements ostore.util.QuickSerializable) for sending error messages
205        
206     - StormBlockMsg class (extends ostore.tapestry.api.TapestryRouteMsg,     - StormBlockMsg class (extends ostore.tapestry.api.TapestryRouteMsg,
207       implements ostore.util.QuickSerializable) to carry a requested Storm       implements ostore.util.QuickSerializable) to carry a requested Storm
208       block being sent to a original requester       block being sent to a original requester
209        
210  For interacting Storm's storage model with Tapestry we must create and event  For interacting Storm's storage model with Tapestry we must create an event
211  handler class:  handler class:
212    
213     - StormTapestryManager class (implements sandStorm.api.EventHandlerIF) for     - StormTapestryManager class (implements sandStorm.api.EventHandlerIF) for
# Line 216  handler class: Line 216  handler class:
216            
217            
218  Here, we propose a simple StormTapestryManager pseudo code. This is an event  Here, we propose a simple StormTapestryManager pseudo code. This is an event
219  handler class for Tapestry services. In addition, this class provides Storm  handler class for Tapestry services. In addition to basic event handling
220  block query and insert operations in a Tapestry overlay.  functionality, this class supports Storm block query and insert operations
221    in a Tapestry overlay.
222    
223          class StormTapestryManager implements EventHandlerIF {          class StormTapestryManager implements EventHandlerIF {
224                    
225                  // dispatcher which dispatch all items                  // Dispatcher which dispatches all items
226                  // to target peer (Mandatory: sandStorm requires this)                  // to a target peer.
227                    // (Mandatory: sandStorm requires this)
228                  method dispatch (QueueElementIF item) {                  method dispatch (QueueElementIF item) {
229                          //try do dispatch an item                          //try do dispatch an item
230                          try {                          try {
# Line 232  block query and insert operations in a T Line 234  block query and insert operations in a T
234                          }                          }
235                  }                  }
236    
237                  // general initialization method                  // General initialization method
238                  // that handles subscribes this stage                  // that handles subscribes this stage
239                  // to listen certain events and messages                  // to listen certain events and messages.
240                  // (Mandatory: sandStorm requires this)                  // (Mandatory: sandStorm requires this)
241                  method init (ConfigDataIF config) {                  method init (ConfigDataIF config) {
242    
243                  // find our NodeId                  // find our NodeId
244                  self_node_id = new NodeId                  self_node_id = new NodeId
245                                    
246                  // initialize the appropriate                  // Initialize the appropriate
247                  // instance of the Classifier. The Classifier handles the                  // instance of the Classifier. The Classifier handles the
248                  // publish/subscribe mechanism used for                  // publish/subscribe mechanism used for
249                  // event dispatch between stages.                  // event dispatch between stages.
250                                    
251                  classifier = Classifier.getClassifier(self_node_id = new NodeId)                  classifier = Classifier.getClassifier(self_node_id = new NodeId)
252                                    
253                  // create peer ID based on peer's public key                  // Create peer ID based on peer's public key.
254                   _peer = new SHA1Hash (pkey);                   peer = new SHA1Hash (publicKey);
255                                    
256                   // the mandatory *events* which we want to listen                   // The *events* which we want to listen.
257                     // (Mandatory: sandStorm requires this)
258    
259                   array event_types = {                   array event_types = {
260                   "seda.sandStorm.api.StagesInitializedSignal",                   "seda.sandStorm.api.StagesInitializedSignal",
# Line 263  block query and insert operations in a T Line 266  block query and insert operations in a T
266                          classifier.subscribe (event_type[i])                          classifier.subscribe (event_type[i])
267                   }                   }
268                                    
269                  // the mandatory *messages* we want receive through Tapestry                  // The *messages* we want receive through Tapestry.
270                    // (Mandatory: sandStorm requires this)
271    
272                  array message_types = {                  array message_types = {
273                  "org.nongnu.storm.p2p.tapestry.StormErrorMsg",                  "org.nongnu.storm.p2p.tapestry.StormErrorMsg",
# Line 277  block query and insert operations in a T Line 281  block query and insert operations in a T
281                  for (From i=0 To numberOf(message_types)) {                  for (From i=0 To numberOf(message_types)) {
282                            
283                     // First, register messages we intend to                     // First, register messages we intend to
284                     // receive                     // receive.
285                                        
286                     TypeTable.register_type(messages_types[i])                     TypeTable.register_type(messages_types[i])
287                                        
288                     // Second, we require that inbound field is                     // Second, require that inbound field is
289                     // set to true.  Otherwise, we will see the messages that we                     // set to true.  Otherwise, we will see the messages that we
290                     // send as well                     // send as well.
291    
292                     classifier.subscribe (message_type[i], verifyInbound);                     classifier.subscribe (message_type[i], verifyInbound);
293                  }                  }
# Line 291  block query and insert operations in a T Line 295  block query and insert operations in a T
295                                    
296                  }                  }
297                                    
298                  // handles *all* events which stage is listening                  // Handles *all* events which stage is listening.
299                  // (Mandatory: sandStorm requires this)                  // (Mandatory: sandStorm requires this)
300                   method handleEvents(QueueElementIF array items)                   method handleEvents(QueueElementIF array items)
301                   throws EventHandlerException {                   throws EventHandlerException {
# Line 300  block query and insert operations in a T Line 304  block query and insert operations in a T
304                          handleEvent(items[i])                                            handleEvent(items[i])                  
305                   }                   }
306                                    
307                  // handles single event for this stage. handleEvents                  // Handles single event for this stage. handleEvents
308                  // method gives an item this method                  // method gives an item to this method as a parameter.
309                  // (Mandatory: sandStorm requires this)                  // (Mandatory: sandStorm requires this)
310                  method handleEvent(QueueElementIF item) {                  method handleEvent(QueueElementIF item) {
311                                    
# Line 309  block query and insert operations in a T Line 313  block query and insert operations in a T
313                          // StagesInitializedSignal received                          // StagesInitializedSignal received
314                            
315                  } else if (item instanceof TapestryReadyMsg) {                  } else if (item instanceof TapestryReadyMsg) {
316                          // connected                                  // Connected to network.            
317                          publishLocalStormPool()                          publishLocalStormPool()
318                  } else if (item instanceof TapestryDetachConfirm) {                  } else if (item instanceof TapestryDetachConfirm) {
319                          // disconnected                          // Disconnected from network.
320                          unpublishLocalStormPool()                          unpublishLocalStormPool()
321                  } else if (item instanceof TapestryLocateFailure) {                  } else if (item instanceof TapestryLocateFailure) {
322                          // Search failed                          // Search failed.                      
                           
323                  } else if (item instanceof StormLocateMsg) {                  } else if (item instanceof StormLocateMsg) {
324                            // Handle StormLocateMsg with a custom method.
325                          handleStormLocateMsg(item)                          handleStormLocateMsg(item)
   
326                  } else if (item instanceof StormResponseMsg) {                  } else if (item instanceof StormResponseMsg) {
327                          handleStormResponseMsg(item)                          // Handle StormResponseMsg with a custom method.
328                                                    handleStormResponseMsg(item)                    
329                  } else if ( item instanceof XXX ) {                  } else if ( item instanceof XXX ) {
330                            // Handle message XXX with a custom method.
331                          handleOtherMsg(item)                          handleOtherMsg(item)
332                          ...                          ...
333                  } else {                  } else {                        
334                          //  Unknown QueueElementIF item                          //  Unknown QueueElementIF item
335                  }                                }              
336                                                                    
337                  }                  }
338                                    
339                  // handles a certain type of event(StormLocateMsg)                  // Handles a certain type of event (StormLocateMsg).
340                  method handleStormLocateMsg(item){                  method handleStormLocateMsg(item){
341                          results = performLocalSearch()                          results = performLocalSearch()
342                          dispatch(results)                                        dispatch(results)              
343                  }                  }
344                                    
345                  // handles a certain type of event(StormResponseMsg)                  // Handles a certain type of event (StormResponseMsg).
346                  method handleStormResponseMsg(item){                  method handleStormResponseMsg(item){
347                          response = createResponse()                          response = createResponse()
348                          dispatch(response)                                        dispatch(response)              
349                  }                  }
350                                    
351                  // handles a certain type of event(XXX)                  // Handles a certain type of event (XXX).
352                  method handleXXXMsg(item){                  method handleXXXMsg(item){
353                          // do something                          // do something
354                  }                  }
355                                                    
356                  // create a Storm query to a Tapestry network                  // Create a Storm query to a Tapestry network,
357                  // Query tag is Storm block's GUID                  // Query tag is Storm block's GUID.
358                  method createQuery(){                  method createQuery(){
359                          blockTag = new BlockIDTag(blockGUID)                          blockTag = new BlockIDTag(blockGUID)
360                          q = new StormQuery()                          q = new StormQuery()
# Line 358  block query and insert operations in a T Line 362  block query and insert operations in a T
362                          dispatch(queryMsg)                          dispatch(queryMsg)
363                  }                  }
364                                    
365                  // dispatch TapestryPublishMsg for all blocks in the                  // Dispatch TapestryPublishMsg for all blocks in a
366                  // local Storm pool                  // local Storm pool.
367                  method publishLocalStormPool(){                          method publishLocalStormPool(){        
368                          for each block in localPool                          for each block in localPool
369                                  block = localPool.getBlock()                                  block = localPool.getBlock()
# Line 369  block query and insert operations in a T Line 373  block query and insert operations in a T
373                                  dispatch(publishMsg)                                              dispatch(publishMsg)            
374                  }                  }
375                                    
376                  // dispatch TapestryUnpublishMsg for all blocks in the                  // Dispatch TapestryUnpublishMsg for all blocks in a
377                  // local Storm pool                  // local Storm pool.
378                  method unpublishLocalStormPool(){                                method unpublishLocalStormPool(){              
379                          for each block in localPool                          for each block in localPool
380                                  block = localPool.getBlock()                                  block = localPool.getBlock()
# Line 380  block query and insert operations in a T Line 384  block query and insert operations in a T
384                                  dispatch(unpublishMsg)                                  dispatch(unpublishMsg)
385                  }                  }
386                                    
387                  // dispatch TapestryPublishMsg for a single block in the                  // Dispatch TapestryPublishMsg for a single block in a
388                  // local Storm pool                              // local Storm pool.            
389                  method publishSingleStormBlock(blockGUID) {                  method publishSingleStormBlock(blockGUID) {
390                          tag = new BlockIDTag(blockGUID)                                                  tag = new BlockIDTag(blockGUID)                        
391                          publishMsg = new                          publishMsg = new
# Line 389  block query and insert operations in a T Line 393  block query and insert operations in a T
393                          dispatch(publishMsg)                          dispatch(publishMsg)
394                  }                  }
395                                    
396                  // dispatch TapestryUnpublishMsg for a single block in the                  // Dispatch TapestryUnpublishMsg for a single block in a
397                  // local Storm pool                              // local Storm pool.            
398                  method unpublishSingleStormBlock(blockGUID) {                  method unpublishSingleStormBlock(blockGUID) {
399                          tag = new BlockIDTag(blockGUID)                                                  tag = new BlockIDTag(blockGUID)                        
400                          unpublishMsg = new                          unpublishMsg = new
# Line 399  block query and insert operations in a T Line 403  block query and insert operations in a T
403                  }                  }
404          }          }
405            
 .. more to come  
406    
407  Changes  Changes
408  =======  =======
409    
410  .. there will be some changes  Eight new classes will be implemented under org.nongnu.storm.p2p.tapestry package
411    into the Storm code base, if this PEG document is accepted.  
412    
413  .. _`Tapestry programmer's guide`: http://www.cs.berkeley.edu/%7Eravenben/tapestry/html/guide.html  .. _`Tapestry programmer's guide`: http://www.cs.berkeley.edu/%7Eravenben/tapestry/html/guide.html
414  .. _Tapestry: http://www.cs.berkeley.edu/~ravenben/publications/pdf/tapestry_jsac.pdf  .. _Tapestry: http://www.cs.berkeley.edu/~ravenben/publications/pdf/tapestry_jsac.pdf

Legend:
Removed from v.1.17  
changed lines
  Added in v.1.18

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26