[jboss-dev-forums] [Design of Messaging on JBoss (Messaging/JBoss)] - Re: XA recovery integration

[juha@jboss.org]

"timfox" wrote : 
  | One thing I'm confused about, is I can see the prepared transaction ids being loaded at server peer startup into the transaction repository and I can see TransactionRepository::getPreparedTransactions returning a list of ids, but I can't see anywhere where the actual transactional state (i.e. the adds/acks) are being loaded from the database and "replayed" into the channel.
  | 

Can you explain what you mean by replayed into the channel in terms of API calls?

The handling of acks and references that I was able to extract from the submitted patch are below. 


  | 	public List getPreparedTransactions()
  |    {
  | 
  | 		ArrayList prepared = new ArrayList();
  | 
  | 		Iterator iter = globalToLocalMap.values().iterator();
  | 
  | 		while (iter.hasNext())
  |       {
  | 			Transaction tx = (Transaction) iter.next();
  | 
  | 			try
  |          {
  | 				if(trace)
  | 					log.trace("Loading and handling refs and acks to the Tx "+tx);
  | 
  |             // TODO: [JPL] should this only apply to STATE_PREPARED transactions?
  | 
  | 				handleReferences(tx, tx.getId());
  | 				handleAcks(tx, tx.getId());
  | 			}
  |          catch (Exception e)
  |          {
  |             // TODO: [JPL] fix this
  | 				e.printStackTrace();
  | 			}
  | 
  | 			if (tx.xid != null && tx.getState() == Transaction.STATE_PREPARED)
  |          {
  | 				prepared.add(tx.getXid());
  | 			}
  | 		}
  | 
  | 		return prepared;
  | 	}
  | 
  | 
  | ...
  | 
  | 
  | /**
  | 	 * Load the references and invoke the channel to handle those refs
  | 	 */
  | 	private void handleReferences(Transaction tx, long txId) throws Exception {
  | 
  | 		long messageId = persistenceManager.getMessageIdForRef(txId);
  | 
  | 		List refsList = getRefs(messageId);
  | 
  | 		// now we got all the refs
  | 		// for each ref loaded, we'll invoke channel.handle
  | 		for (Iterator iter = refsList.iterator(); iter.hasNext();)
  | 		{
  | 			CoreDestination d = getChannel(persistenceManager.getChannelId(txId), txId);
  | 
  |          if (trace)
  | 			   log.trace("Handling the channel");
  | 
  |          d.handle(null, (MessageReference) iter.next(), tx);
  | 		}
  | 	}
  | 
  | 	/**
  | 	 * Load the acks and acknowledge them
  | 	 */
  | 	private void handleAcks(Transaction tx, long txId) throws Exception {
  | 
  | 		long messageId = persistenceManager.getMessageIdForAck(txId);
  | 
  | 		List refsList = getRefs(messageId);
  | 
  | 		for (Iterator iter = refsList.iterator(); iter.hasNext();)
  | 		{
  | 			Delivery del = new SimpleDelivery(null, (MessageReference) iter.next());
  | 
  | 			try
  |          {
  | 				if(trace)
  | 					log.trace("Acknowledging..");
  | 
  | 				((DeliveryObserver)del).acknowledge(del, tx);
  | 			}
  |          catch (Throwable e)
  |          {
  |             // TODO: [JPL] fix this
  | 				e.printStackTrace();
  | 			}
  | 		}
  | 	}
  | 
  | 	/**
  | 	 * Get the message references based on the messageId from database
  | 	 */
  | 	private List getRefs(long messageId) throws Exception
  | 	{
  | 		List noRefsList = new ArrayList();
  | 		List refsList = new ArrayList();
  | 
  | 		// Find the message store
  |       // TODO: [JPL] this needs to be fixed to go through the kernel
  | 		MessageStore ms = getMessageStore();
  | 
  | 		// and message reference from store
  | 		MessageReference ref = ms.reference(messageId);
  | 
  | 		// Store, sometime, does'nt know about the message referece
  | 		// and the above ref may be null. Hence we need to load actual message
  | 		// by goind back to the database and loading them based on id
  | 
  | 		if (ref == null)
  | 		{
  | 			noRefsList.add(new Long(messageId));
  | 		}
  | 		else
  | 		{
  | 			refsList.add(ref);
  | 		}
  | 
  | 		// ask the pm to get the messages from messageId list
  | 		List messagesList = persistenceManager.getMessages(noRefsList);
  | 
  | 		for (Iterator iter = messagesList.iterator(); iter.hasNext();)
  | 		{
  | 			Message m = (Message) iter.next();
  | 			MessageReference r = ms.reference(m);
  | 			refsList.add(r);
  | 		}
  | 
  | 		return refsList;
  | 	}
  | 
  | 
  | 


"timfox" wrote : 
  | 
  | I can also see the code Madhu supplied to do this has been removed. What am I missing here?
  | 
  | 

There should be no code removed. The only thing is some additions may have accidentally been omitted due to trying to extract a working patch from the multiple submissions. Let me know if something obvious is missing.


View the original post : http://www.jboss.com/index.html?module=bb&op=viewtopic&p=3988227#3988227

Reply to the post : http://www.jboss.com/index.html?module=bb&op=posting&mode=reply&p=3988227