Java > Open Source Codes > org > opencrx > kernel > workflow > servlet > SubscriptionHandlerServlet


1 /*
2  * ====================================================================
3  * Project: opencrx, http://www.opencrx.org/
4  * Name: $Id: SubscriptionHandlerServlet.java,v 1.9 2006/04/02 00:34:25 wfro Exp $
5  * Description: openCRX EventHandler
6  * Revision: $Revision: 1.9 $
7  * Owner: CRIXP AG, Switzerland, http://www.crixp.com
8  * Date: $Date: 2006/04/02 00:34:25 $
9  * ====================================================================
10  *
11  * This software is published under the BSD license
12  * as listed below.
13  *
14  * Copyright (c) 2004-2005, CRIXP Corp., Switzerland
15  * All rights reserved.
16  *
17  * Redistribution and use in source and binary forms, with or without
18  * modification, are permitted provided that the following conditions
19  * are met:
20  *
21  * * Redistributions of source code must retain the above copyright
22  * notice, this list of conditions and the following disclaimer.
23  *
24  * * Redistributions in binary form must reproduce the above copyright
25  * notice, this list of conditions and the following disclaimer in
26  * the documentation and/or other materials provided with the
27  * distribution.
28  *
29  * * Neither the name of CRIXP Corp. nor the names of the contributors
30  * to openCRX may be used to endorse or promote products derived
31  * from this software without specific prior written permission
32  *
33  *
34  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
35  * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
36  * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
37  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
38  * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS
39  * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
40  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
41  * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
42  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
43  * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
44  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
45  * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
46  * POSSIBILITY OF SUCH DAMAGE.
47  *
48  * ------------------
49  *
50  * This product includes software developed by the Apache Software
51  * Foundation (http://www.apache.org/).
52  *
53  * This product includes software developed by contributors to
54  * openMDX (http://www.openmdx.org/)
55  */
56 package org.opencrx.kernel.workflow.servlet;
57
58 import java.io.IOException ;
59 import java.io.UnsupportedEncodingException ;
60 import java.security.MessageDigest ;
61 import java.security.NoSuchAlgorithmException ;
62 import java.util.ArrayList ;
63 import java.util.Date ;
64 import java.util.Iterator ;
65 import java.util.List ;
66
67 import javax.jmi.reflect.RefException;
68 import javax.naming.Context ;
69 import javax.naming.InitialContext ;
70 import javax.naming.NamingException ;
71 import javax.servlet.ServletConfig ;
72 import javax.servlet.ServletException ;
73 import javax.servlet.http.HttpServlet ;
74 import javax.servlet.http.HttpServletRequest ;
75 import javax.servlet.http.HttpServletResponse ;
76
77 import org.opencrx.kernel.base.cci.AuditEntry;
78 import org.opencrx.kernel.base.cci.AuditEntryFilter;
79 import org.opencrx.kernel.base.cci.Auditee;
80 import org.opencrx.kernel.base.cci.ExecuteWorkflowParams;
81 import org.opencrx.kernel.base.cci.ObjectCreationAuditEntry;
82 import org.opencrx.kernel.base.cci.ObjectModificationAuditEntry;
83 import org.opencrx.kernel.base.cci.ObjectRemovalAuditEntry;
84 import org.opencrx.kernel.base.cci.TestAndSetVisitedByResult;
85 import org.opencrx.kernel.home1.cci.Subscription;
86 import org.opencrx.kernel.home1.cci.UserHome;
87 import org.opencrx.kernel.workflow1.cci.Topic;
88 import org.opencrx.kernel.workflow1.cci.WfProcess;
89 import org.openmdx.application.log.AppLog;
90 import org.openmdx.base.accessor.generic.view.Manager_1;
91 import org.openmdx.base.accessor.jmi.cci.JmiServiceException;
92 import org.openmdx.base.accessor.jmi.cci.RefContainer_1_0;
93 import org.openmdx.base.accessor.jmi.cci.RefPackage_1_0;
94 import org.openmdx.base.accessor.jmi.spi.RefRootPackage_1;
95 import org.openmdx.base.exception.ServiceException;
96 import org.openmdx.base.text.conversion.Base64;
97 import org.openmdx.compatibility.application.dataprovider.transport.ejb.cci.Dataprovider_1ConnectionFactoryImpl;
98 import org.openmdx.compatibility.base.dataprovider.cci.DataproviderOperations;
99 import org.openmdx.compatibility.base.dataprovider.cci.QualityOfService;
100 import org.openmdx.compatibility.base.dataprovider.cci.RequestCollection;
101 import org.openmdx.compatibility.base.dataprovider.cci.ServiceHeader;
102 import org.openmdx.compatibility.base.dataprovider.transport.adapter.Provider_1;
103 import org.openmdx.compatibility.base.dataprovider.transport.cci.Dataprovider_1_1Connection;
104 import org.openmdx.compatibility.base.dataprovider.transport.cci.Provider_1_0;
105 import org.openmdx.compatibility.base.dataprovider.transport.delegation.Connection_1;
106 import org.openmdx.compatibility.base.naming.Path;
107 import org.openmdx.compatibility.base.query.FilterOperators;
108 import org.openmdx.compatibility.base.query.FilterProperty;
109 import org.openmdx.compatibility.base.query.Quantors;
110 import org.openmdx.model1.accessor.basic.spi.Model_1;
111
112 /**
113  * The SubscriptionHandlerServlet 'listens' for object modifications on incoming
114  * audit entries. For each active subscription (i.e. topic target
115  * pattern) which's filter (object identity and attribute filters) matches
116  * the modified object, the corresponding workflows are executed.
117  */
118 public class SubscriptionHandlerServlet
119     extends HttpServlet {
120
121     //-----------------------------------------------------------------------
122 public void init(
123         ServletConfig config
124     ) throws ServletException {
125
126         super.init(config);
127         
128         // initialize model repository
129 int i = 0;
130         List modelPackages = new ArrayList ();
131         while(getInitParameter("modelPackage[" + i + "]") != null) {
132             modelPackages.add(
133               getInitParameter("modelPackage[" + i + "]")
134             );
135             i++;
136         }
137         try {
138             new Model_1().addModels(modelPackages);
139         }
140         catch(Exception e) {
141             System.out.println("can not initialize model repository " + e.getMessage());
142             System.out.println(new ServiceException(e).getCause());
143         }
144
145         Context initialContext;
146         try {
147             initialContext = new InitialContext ();
148         } catch (NamingException e) {
149             throw new ServletException ("can not get the initial context", e);
150         }
151       
152         // data connection
153 try {
154             this.connectionData = Dataprovider_1ConnectionFactoryImpl.createGenericConnection(
155                 initialContext.lookup("java:comp/env/ejb/data")
156             );
157         }
158         catch(Exception e) {
159             throw new ServletException ("can not get connection to data provider", e);
160         }
161
162     }
163
164     //-----------------------------------------------------------------------
165 private Short getEventType(
166         AuditEntry auditEntry
167     ) {
168         return new Short (
169             auditEntry instanceof ObjectRemovalAuditEntry
170                 ? DataproviderOperations.OBJECT_REMOVAL
171                 : auditEntry instanceof ObjectCreationAuditEntry
172                     ? DataproviderOperations.OBJECT_CREATION
173                     : auditEntry instanceof ObjectModificationAuditEntry
174                         ? DataproviderOperations.OBJECT_REPLACEMENT
175                         : 0
176         );
177     }
178
179     //-----------------------------------------------------------------------
180 private boolean subscriptionAcceptsEventType(
181         Subscription subscription,
182         Short eventType
183     ) {
184         if(
185             (subscription.getEventType() == null) ||
186             (subscription.getEventType().size() == 0)
187         ) {
188             return true;
189         }
190         for(
191             Iterator i = subscription.getEventType().iterator();
192             i.hasNext();
193         ) {
194             Number e = (Number )i.next();
195             if(eventType.compareTo(e) == 0) {
196                 return true;
197             }
198         }
199         return false;
200     }
201     
202     //-----------------------------------------------------------------------
203 public boolean subscriptionAcceptsMessage(
204         Subscription subscription,
205         AuditEntry auditEntry
206     ) {
207         // verify active and event type
208 Short eventType = this.getEventType(auditEntry);
209         if(!subscription.isActive() || !this.subscriptionAcceptsEventType(subscription, eventType)) {
210             return false;
211         }
212
213         // If the message is a UserHome object, assert that the owner
214 // of the message and subscription are identical. This way users
215 // can not subscribe to UserHome objects of other users.
216 Path auditEntryPath = new Path(auditEntry.refMofId());
217         Path userHomeIdentity = new Path(subscription.refMofId()).getParent().getParent();
218         if(auditEntryPath.getPrefix(7).isLike(USER_HOMES)) {
219             if(!auditEntryPath.startsWith(userHomeIdentity)) {
220                 return false;
221             }
222         }
223         
224         boolean accepts = true;
225         // TODO: verify filter
226 // for(int i = 0; i < MAX_ATTRIBUTE_FILTERS; i++) {
227 // if(this.subscription.values("filterName" + i).size() > 0) {
228 // String filterValue = (String)this.subscription.values("filterValue" + i).get(0);
229 // Object messageValue = auditEntry.values(
230 // (String)this.subscription.values("filterName" + i).get(0)
231 // ).get(0);
232 // accepts = (filterValue == null) || (messageValue == null)
233 // ? messageValue == filterValue
234 // : messageValue.equals(filterValue);
235 // if(!accepts) break;
236 // }
237 // }
238 return accepts;
239     }
240         
241     //-----------------------------------------------------------------------
242 public boolean topicAcceptsObjectXri(
243         String providerName,
244         String segmentName,
245         Topic topic,
246         String objectXri
247     ) {
248         String topicPatternXri = topic.getTopicPathPattern();
249         if(topicPatternXri != null) {
250             Path topicPattern = new Path(topicPatternXri);
251             Path objectPath = new Path(objectXri);
252             return topicPattern.size() < 7
253                 ? false
254                 : objectPath.isLike(topicPattern) &&
255                   objectPath.get(2).equals(providerName) &&
256                   objectPath.get(4).equals(segmentName);
257         }
258         else {
259             return false;
260         }
261     }
262     
263     //-----------------------------------------------------------------------
264 private List findSubscriptions(
265         String providerName,
266         String segmentName,
267         org.opencrx.kernel.workflow1.cci.Segment workflowSegment,
268         org.opencrx.kernel.home1.cci.Segment userHomeSegment,
269         AuditEntry auditEntry
270     ) throws ServiceException {
271         // Find topics matching the auditee (modified, created or removed object)
272 List topics = new ArrayList ();
273         for(
274             Iterator i = workflowSegment.getTopic().iterator();
275             i.hasNext();
276         ) {
277             Topic t = (Topic)i.next();
278             if(this.topicAcceptsObjectXri(providerName, segmentName, t, auditEntry.getAuditee())) {
279                 topics.add(t);
280             }
281         }
282         // Find subscriptions for this topic
283 List subscriptions = null;
284         if(topics.size() > 0) {
285             subscriptions = new ArrayList ();
286             for(
287                 Iterator i = topics.iterator();
288                 i.hasNext();
289             ) {
290                 Topic topic = (Topic)i.next();
291                 FilterProperty[] filter =
292                     new FilterProperty[]{
293                         new FilterProperty(
294                             Quantors.THERE_EXISTS,
295                             "identity",
296                             FilterOperators.IS_LIKE,
297                             new String []{
298                                 new Path("xri:@openmdx:org.opencrx.kernel.home1/provider/" + providerName + "/segment/" + segmentName + "/userHome/:*/subscription/:*").toXri()
299                             }
300                         ),
301                         new FilterProperty(
302                             Quantors.THERE_EXISTS,
303                             "topic",
304                             FilterOperators.IS_IN,
305                             new Path[]{new Path(topic.refMofId())}
306                         )
307                     };
308                 for(
309                     Iterator j = ((RefContainer_1_0)userHomeSegment.getExtent()).subSet(filter).iterator();
310                     j.hasNext();
311                 ) {
312                     Subscription subscription = (Subscription)j.next();
313                     if(this.subscriptionAcceptsMessage(subscription, auditEntry)) {
314                         subscriptions.add(subscription);
315                     }
316                 }
317             }
318         }
319         return subscriptions;
320     }
321
322     //-----------------------------------------------------------------------
323 private void handleSubscriptions(
324         String providerName,
325         String segmentName,
326         RefPackage_1_0 rootPkg,
327         org.opencrx.kernel.workflow1.cci.Segment workflowSegment,
328         org.opencrx.kernel.home1.cci.Segment userHomeSegment,
329         org.opencrx.kernel.base.cci.basePackage basePkg,
330         List auditEntries
331     ) throws ServiceException {
332         
333         int ii = 0;
334         for(
335             Iterator i = auditEntries.iterator();
336             (ii < 100) && i.hasNext();
337             ii++
338         ) {
339             AuditEntry auditEntry = (AuditEntry)i.next();
340             TestAndSetVisitedByResult markAsVisistedReply = null;
341             try {
342                 markAsVisistedReply = auditEntry.testAndSetVisitedBy(
343                     VISITOR_ID
344                 );
345             }
346             catch(RefException e) {
347                 AppLog.info("Can not invoke markAsVisited", e.getMessage());
348             }
349             if(
350                 (markAsVisistedReply != null) ||
351                 (markAsVisistedReply.getVisitStatus() == 0)
352             ) {
353                 List subscriptions = this.findSubscriptions(
354                     providerName,
355                     segmentName,
356                     workflowSegment,
357                     userHomeSegment,
358                     auditEntry
359                 );
360                 if(subscriptions != null) {
361                     for(
362                         Iterator j = subscriptions.iterator();
363                         j.hasNext();
364                     ) {
365                         Subscription subscription = (Subscription)j.next();
366                         UserHome userHome = (UserHome)rootPkg.refObject(
367                             new Path(subscription.refMofId()).getParent().getParent().toXri()
368                         );
369                         
370                         // Execute all actions attached to the topic
371 for(
372                             Iterator k = subscription.getTopic().getPerformAction().iterator();
373                             k.hasNext();
374                         ) {
375                             WfProcess performAction = (WfProcess)k.next();
376                             
377                             // Execute workflow
378 // Synchronous workflows are executed immediately. For asynchronous
379 // workflows a WfProcessInstance is created. The eventId is constructed
380 // from the ids of the action and audit entry.
381 try {
382                                 MessageDigest md = MessageDigest.getInstance("MD5");
383                                 md.update(auditEntry.refMofId().getBytes("UTF-8"));
384                                 md.update(performAction.refMofId().getBytes("UTF-8"));
385                                 ExecuteWorkflowParams params = basePkg.createExecuteWorkflowParams(
386                                     null,
387                                     null,
388                                     auditEntry.getAuditee(),
389                                     Base64.encode(md.digest()).replace('/', '-'),
390                                     new Integer (this.getEventType(auditEntry).intValue()),
391                                     subscription,
392                                     performAction
393                                 );
394                                 try {
395                                     rootPkg.refBegin();
396                                     userHome.executeWorkflow(params);
397                                     rootPkg.refCommit();
398                                 }
399                                 catch(RefException e) {
400                                     AppLog.warning("Can not execute workflow", performAction.getName() + "; home=" + userHome.refMofId());
401                                     AppLog.warning(e.getMessage(), e.getCause(), 1);
402                                     try {
403                                         rootPkg.refRollback();
404                                     } catch(Exception e0) {}
405                                 }
406                             }
407                             catch(NoSuchAlgorithmException e) {
408                                 new ServiceException(e).log();
409                             }
410                             catch (UnsupportedEncodingException e) {
411                                 new ServiceException(e).log();
412                             }
413                         }
414                     }
415                 }
416             }
417         }
418     }
419     
420     //-------------------------------------------------------------------------
421 private RefPackage_1_0 createDataPkg(
422         ServiceHeader header
423     ) throws ServiceException {
424         Provider_1_0 provider = new Provider_1(
425           new RequestCollection(
426             header,
427             this.connectionData
428           ),
429           false
430         );
431         Manager_1 manager = new Manager_1(
432           new Connection_1(
433             provider,
434             false
435           )
436         );
437         RefPackage_1_0 rootPkg = new RefRootPackage_1(manager);
438         return rootPkg;
439     }
440
441     //-----------------------------------------------------------------------
442 public void handleSubscriptions(
443         String id,
444         String providerName,
445         String segmentName,
446         HttpServletRequest req,
447         HttpServletResponse res
448     ) throws IOException {
449         
450         System.out.println(new Date ().toString() + ": openCRX/SubscriptionHandler: " + providerName + "/" + segmentName);
451
452         try {
453             RefPackage_1_0 rootPkg = this.createDataPkg(
454                 new ServiceHeader("admin-" + segmentName, null, false, new QualityOfService())
455             );
456             WorkflowControllerServlet.initWorkflows(
457                 rootPkg,
458                 providerName,
459                 segmentName
460             );
461             
462             // Get auditees
463 List auditSegments = new ArrayList ();
464             auditSegments.add(
465                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.account1/provider/" + providerName + "/segment/" + segmentName).toXri())
466             );
467             auditSegments.add(
468                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.activity1/provider/" + providerName + "/segment/" + segmentName).toXri())
469             );
470             auditSegments.add(
471                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.building1/provider/" + providerName + "/segment/" + segmentName).toXri())
472             );
473             auditSegments.add(
474                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.contract1/provider/" + providerName + "/segment/" + segmentName).toXri())
475             );
476             auditSegments.add(
477                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.depot1/provider/" + providerName + "/segment/" + segmentName).toXri())
478             );
479             auditSegments.add(
480                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.document1/provider/" + providerName + "/segment/" + segmentName).toXri())
481             );
482             auditSegments.add(
483                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.forecast1/provider/" + providerName + "/segment/" + segmentName).toXri())
484             );
485             auditSegments.add(
486                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.model1/provider/" + providerName + "/segment/" + segmentName).toXri())
487             );
488             auditSegments.add(
489                 rootPkg.refObject(new Path("xri:@openmdx:org.opencrx.kernel.product1/provider/" + providerName + "/segment/" + segmentName).toXri())
490             );
491                                 
492             // Workflow segment
493 org.opencrx.kernel.workflow1.cci.Segment workflowSegment = (org.opencrx.kernel.workflow1.cci.Segment)rootPkg.refObject(
494                 new Path("xri:@openmdx:org.opencrx.kernel.workflow1/provider/" + providerName + "/segment/" + segmentName).toXri()
495             );
496             // User home segment
497 org.opencrx.kernel.home1.cci.Segment userHomeSegment = (org.opencrx.kernel.home1.cci.Segment)rootPkg.refObject(
498                 new Path("xri:@openmdx:org.opencrx.kernel.home1/provider/" + providerName + "/segment/" + segmentName).toXri()
499             );
500             // base package
501 org.opencrx.kernel.base.cci.basePackage basePkg = (org.opencrx.kernel.base.cci.basePackage)rootPkg.refPackage(
502                 org.opencrx.kernel.base.cci.basePackage.class.getName()
503             );
504             
505             // Iterate all auditees and check for new audit entries
506 for(
507                 Iterator j = auditSegments.iterator();
508                 j.hasNext();
509             ) {
510                 Auditee auditee = (Auditee)j.next();
511                 AuditEntryFilter filter = basePkg.createAuditEntryFilter();
512                 filter.forAllVisitedBy(
513                     FilterOperators.IS_NOT_IN,
514                     new String []{VISITOR_ID}
515                 );
516                 try {
517                     this.handleSubscriptions(
518                         providerName,
519                         segmentName,
520                         rootPkg,
521                         workflowSegment,
522                         userHomeSegment,
523                         basePkg,
524                         auditee.getAudit(filter)
525                     );
526                 }
527                 catch(ServiceException e) {
528                     System.out.println(new Date () + ": openCRX/SubscriptionHandler: exception occured " + e.getMessage() + ". Continuing");
529                 }
530             }
531         }
532         catch(ServiceException e) {
533             System.out.println(new Date () + ": openCRX/SubscriptionHandler: exception occured " + e.getMessage() + ". Continuing");
534             new ServiceException(e).log();
535         }
536         catch(JmiServiceException e) {
537             System.out.println(new Date () + ": openCRX/SubscriptionHandler: exception occured " + e.getMessage() + ". Continuing");
538             new ServiceException(e).log();
539         }
540         catch(Exception e) {
541             System.out.println(new Date () + ": openCRX/SubscriptionHandler: exception occured " + e.getMessage() + ". Continuing");
542             new ServiceException(e).log();
543         }
544     }
545     
546     //-----------------------------------------------------------------------
547 protected void handleRequest(
548         HttpServletRequest req,
549         HttpServletResponse res
550     ) throws ServletException , IOException {
551         String segmentName = req.getParameter("segment");
552         String providerName = req.getParameter("provider");
553         String id = providerName + "/" + segmentName;
554         // run
555 if(
556             COMMAND_EXECUTE.equals(req.getPathInfo()) &&
557             !this.runningSegments.contains(id)
558         ) {
559             try {
560                 this.runningSegments.add(id);
561                 this.handleSubscriptions(
562                     id,
563                     providerName,
564                     segmentName,
565                     req,
566                     res
567                 );
568             } catch(Exception e) {
569                 new ServiceException(e).log();
570             }
571             finally {
572                 this.runningSegments.remove(id);
573             }
574         }
575     }
576
577     //-----------------------------------------------------------------------
578 protected void doGet(
579         HttpServletRequest req,
580         HttpServletResponse res
581     ) throws ServletException , IOException {
582         res.setStatus(HttpServletResponse.SC_OK);
583         res.flushBuffer();
584         this.handleRequest(
585             req,
586             res
587         );
588     }
589         
590     //-----------------------------------------------------------------------
591 protected void doPost(
592         HttpServletRequest req,
593         HttpServletResponse res
594     ) throws ServletException , IOException {
595         res.setStatus(HttpServletResponse.SC_OK);
596         res.flushBuffer();
597         this.handleRequest(
598             req,
599             res
600         );
601     }
602         
603     //-----------------------------------------------------------------------
604 // Members
605 //-----------------------------------------------------------------------
606 private static final long serialVersionUID = 7074135054692868453L;
607
608     private static final Path USER_HOMES = new Path("xri:@openmdx:org.opencrx.kernel.home1/provider/:*/segment/:*/userHome/:*");
609     private static final String COMMAND_EXECUTE = "/execute";
610     private static final String VISITOR_ID = "SubscriptionHandler";
611
612     private Dataprovider_1_1Connection connectionData = null;
613     private final List runningSegments = new ArrayList ();
614         
615 }
616
617 //--- End of File -----------------------------------------------------------
618

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags