Actually that is what I had tried initially. I added an interceptor at the end and used a flag on the message to decide on reprocessing. But it didn't work after 2 and half days of effort so I gave in to modifying cxf code.
It would be great if some of the interceptors could "undo" the change they did to the message. Then, we would be able to move the message up and down certain portions of the chain. (It might make the chain heavier or slow down processing, but we could always add an annotation or configuration to put "undo" on/off.) Combined with the pause/resume functionality, this would open up some interesting possibilities. Regards Mayank -----Original Message----- From: Daniel Kulp [mailto:[EMAIL PROTECTED] Sent: Wednesday, January 16, 2008 02:14 To: [email protected] Cc: Mayank Thakore Subject: Re: ABR Communication Pattern Hmm... interesting solution. I like it. :-) One potential issue is the use of the ThreadLocal, but that's definitely not an issue right now. The API's technically allow an interceptor to "pause" the chain and then resume it on a separate thread. (for example, to wait for a resource to become available) Nothing does that right now and doing so is going to take a bunch of testing to fix everything else that will break. :-) To "support" that, the variable would need to be stored as a property on the Exchange. Then an interceptor would need to be added to the "end" of the outgoing chains to check for the flag and, if present, restart the message processing from the beginning. I guess that would be the more "correct" way to do it, but also a LOT more complex. For now, your solution is great. Dan On Tuesday 15 January 2008, Mayank Thakore wrote: > I LOVE OPEN SOURCE > > It worked :) > > just need to call following from service implementation when another > response needs to be sent (as long as you keep calling this your > method will get reinvoked to handle this incoming request; so you can > send as many responses to each request as you want) > > JMSDestination.abrStatus.set(new Boolean(true)); > > > > modified code from JMSDestination: > ---------------------------------------------------------------------- >------------------- > > public static ThreadLocal<Boolean> abrStatus = new > ThreadLocal<Boolean>() { @Override > protected synchronized Boolean initialValue() { > return new Boolean(false); > } > }; > > protected class JMSExecutor implements Runnable { > javax.jms.Message message; > > JMSExecutor(javax.jms.Message m) { > message = m; > } > > public void run() { > long abrCount = 0; > do { > getLogger().log(Level.INFO, "run the incoming message > in the threadpool"); > JMSDestination.abrStatus.set(new Boolean(false)); > getLogger().log(Level.INFO, "abr count for > message("+Thread.currentThread()+"): "+abrCount); > abrCount++; > try { > incoming(message); > } catch (IOException ex) { > // TODO: Decide what to do if we receive the > exception. getLogger().log(Level.WARNING, "Failed to process incoming > message : ", ex); > break; > } > } while(JMSDestination.abrStatus.get().booleanValue()); > } > > } > > > ---------------------------------------------------------------------- >------------------- > > Regards > Mayank > > On 1/15/08, Mayank Thakore <[EMAIL PROTECTED]> wrote: > > Well, it didn't work. The interceptors mutate the message and won't > > process the mutated one. > > > > So I am going to try changing the JMSDestination.incoming code to > > resend the message as many times as required. > > > > Any thoughts? > > > > Regards > > Mayank > > > > -----Original Message----- > > From: Mayank Thakore [mailto:[EMAIL PROTECTED] > > Sent: Sunday, January 13, 2008 23:09 > > To: [email protected] > > Subject: ABR Communication Pattern > > > > Hi, > > > > I was trying to achieve Asynchronous Batch Response (ABR) pattern > > with CXF for JMS transport. ABR means user can send multiple > > responses to a single request. > > > > Please read this and let me know what you think. > > > > Hazard info: I haven't finished the client for this yet, so don't > > know wheather it is working. It does print the cxf logs correctly. > > > > Below is my interceptor code. It attaches to the end of the > > interceptor chain and executes the previous two interceptors which > > invoke the service implementation and send the out message > > respectively. > > ====================================================== > > > > package ws.v1.tmf854; > > > > import java.util.ListIterator; > > > > import org.apache.cxf.interceptor.Fault; > > import org.apache.cxf.interceptor.Interceptor; > > import org.apache.cxf.message.Message; > > import org.apache.cxf.phase.AbstractPhaseInterceptor; > > import org.apache.cxf.phase.Phase; > > > > public class ABRInterceptor extends > > AbstractPhaseInterceptor<Message> { > > > > public ABRInterceptor() { > > super(Phase.POST_INVOKE); > > getAfter().add( > > > > org.apache.cxf.interceptor.OutgoingChainInterceptor.class > > .getName()); > > } > > > > @Override > > public void handleMessage(Message message) throws Fault { > > System.out.println("ABRInterceptor invoked"); > > if (ABRSession.isEnabled()) { > > executeReRuns(message); > > } > > } > > > > private void executeReRuns(Message message) { > > ListIterator<Interceptor<? extends Message>> iterator > > = prepareReRunIterator(message); > > while (ABRSession.isEnabled()) { > > ABRSession.disable(); > > for (int i = 0; i < 2; i++) { > > iterator.previous(); > > } > > for (int i = 0; i < 2; i++) { > > Interceptor currentInterceptor = > > iterator.next(); > > > > currentInterceptor.handleMessage(message); } > > } > > } > > > > private ListIterator<Interceptor<? extends Message>> > > prepareReRunIterator( > > Message message) { > > ListIterator<Interceptor<? extends Message>> iterator > > = message > > .getInterceptorChain().getIterator(); > > while (iterator.hasNext()) { > > iterator.next(); > > } > > iterator.previous(); > > return iterator; > > } > > > > } > > > > > > > > ====================================================== > > > > > > > > Below is the session control device. It uses a thread local variable > > to remember if ABR session is enabled or disabled. > > ====================================================== > > > > package ws.v1.tmf854; > > > > public class ABRSession { > > > > private static ThreadLocal<Boolean> status = new > > ThreadLocal<Boolean>() { > > @Override > > protected synchronized Boolean initialValue() { > > return new Boolean(false); > > } > > }; > > > > public static synchronized boolean isEnabled() { > > return status.get().booleanValue(); > > } > > > > public static synchronized void disable() { > > System.out.println("Disabling ABRSession"); > > status.set(new Boolean(false)); > > > > } > > > > public static synchronized void enable() { > > System.out.println("Enabling ABRSession"); > > status.set(new Boolean(true)); > > > > } > > > > } > > > > > > > > ====================================================== > > > > So, if user wants ABR mode, they just need to invoke > > ABRSession.enable() inside the service implementation. > > > > Here is the server main for completion: > > ====================================================== > > > > package ws.v1.tmf854; > > > > import javax.xml.ws.Endpoint; > > > > import org.apache.cxf.jaxws.EndpointImpl; > > > > /** > > * This class was generated by Apache CXF (incubator) > > 2.0.3-incubator Sat Jan 12 > > * 11:10:39 IST 2008 Generated source version: 2.0.3-incubator > > * > > */ > > > > public class AlarmRetrieval_AlarmRetrievalJms_Server { > > > > protected AlarmRetrieval_AlarmRetrievalJms_Server() throws > > Exception { > > System.out.println("Starting Server"); > > Object implementor = new AlarmRetrievalImpl(); > > String address = "jms://"; > > EndpointImpl ep = (EndpointImpl) > > Endpoint.publish(address, implementor); > > ep.getServer().getEndpoint().getInInterceptors().add( > > new ABRInterceptor()); > > } > > > > public static void main(String args[]) throws Exception { > > new AlarmRetrieval_AlarmRetrievalJms_Server(); > > System.out.println("Server ready..."); > > > > } > > } > > > > > > ========================================================= > > > > > > So, what do you think? > > > > Thanks for any and all comments. Feel free to be critical. > > > > Mayank -- J. Daniel Kulp Principal Engineer, IONA [EMAIL PROTECTED] http://www.dankulp.com/blog
