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 |
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 |
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 |
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 { |
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", |
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", |
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 |
} |
} |
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 { |
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 |
|
|
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() |
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() |
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() |
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 |
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 |
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 |