Save This Page
Home » activemq-parent-5.3.1-source-release » org.apache » activemq » store » kahadaptor » [javadoc | source]
    1   /**
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  You may obtain a copy of the License at
    8    *
    9    *      http://www.apache.org/licenses/LICENSE-2.0
   10    *
   11    * Unless required by applicable law or agreed to in writing, software
   12    * distributed under the License is distributed on an "AS IS" BASIS,
   13    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   14    * See the License for the specific language governing permissions and
   15    * limitations under the License.
   16    */
   17   package org.apache.activemq.store.kahadaptor;
   18   
   19   import java.io.IOException;
   20   import java.util.ArrayList;
   21   import java.util.List;
   22   
   23   import org.apache.activemq.command.BaseCommand;
   24   import org.apache.activemq.command.Message;
   25   import org.apache.activemq.command.MessageAck;
   26   import org.apache.activemq.store.MessageStore;
   27   import org.apache.commons.logging.Log;
   28   import org.apache.commons.logging.LogFactory;
   29   
   30   /**
   31    * Stores a messages/acknowledgements for a transaction
   32    * 
   33    * @version $Revision: 1.4 $
   34    */
   35   class KahaTransaction {
   36       protected List<TxCommand> list = new ArrayList<TxCommand>();
   37   
   38       void add(KahaMessageStore store, BaseCommand command) {
   39           TxCommand tx = new TxCommand();
   40           tx.setCommand(command);
   41           tx.setMessageStoreKey(store.getId());
   42           list.add(tx);
   43       }
   44   
   45       Message[] getMessages() {
   46           List<BaseCommand> result = new ArrayList<BaseCommand>();
   47           for (int i = 0; i < list.size(); i++) {
   48               TxCommand command = list.get(i);
   49               if (command.isAdd()) {
   50                   result.add(command.getCommand());
   51               }
   52           }
   53           Message[] messages = new Message[result.size()];
   54           return result.toArray(messages);
   55       }
   56   
   57       MessageAck[] getAcks() {
   58           List<BaseCommand> result = new ArrayList<BaseCommand>();
   59           for (int i = 0; i < list.size(); i++) {
   60               TxCommand command = list.get(i);
   61               if (command.isRemove()) {
   62                   result.add(command.getCommand());
   63               }
   64           }
   65           MessageAck[] acks = new MessageAck[result.size()];
   66           return result.toArray(acks);
   67       }
   68   
   69       void prepare() {
   70       }
   71   
   72       void rollback() {
   73           list.clear();
   74       }
   75   
   76       /**
   77        * @throws IOException
   78        */
   79       void commit(KahaTransactionStore transactionStore) throws IOException {
   80           for (int i = 0; i < list.size(); i++) {
   81               TxCommand command = list.get(i);
   82               MessageStore ms = transactionStore.getStoreById(command.getMessageStoreKey());
   83               if (command.isAdd()) {
   84                   ms.addMessage(null, (Message)command.getCommand());
   85               }
   86           }
   87           for (int i = 0; i < list.size(); i++) {
   88               TxCommand command = list.get(i);
   89               MessageStore ms = transactionStore.getStoreById(command.getMessageStoreKey());
   90               if (command.isRemove()) {
   91                   ms.removeMessage(null, (MessageAck)command.getCommand());
   92               }
   93           }
   94       }
   95   
   96       List<TxCommand> getList() {
   97           return new ArrayList<TxCommand>(list);
   98       }
   99   
  100       void setList(List<TxCommand> list) {
  101           this.list = list;
  102       }
  103   }

Save This Page
Home » activemq-parent-5.3.1-source-release » org.apache » activemq » store » kahadaptor » [javadoc | source]