gt:                     pt = pausable.getPausedTransaction();
context: /*
   * jPOS Project [http://jpos.org]
   * Copyright (C) 2000-2015 Alejandro P. Revilla
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Affero General Public License as
   * published by the Free Software Foundation, either version 3 of the
   * License, or (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with this program.  If not, see <http://www.gnu.org/licenses/>.
   */
  
  package org.jpos.transaction;
  
  import org.jdom.Element;
  import org.jpos.core.Configuration;
  import org.jpos.core.ConfigurationException;
  import org.jpos.q2.QBeanSupport;
  import org.jpos.q2.QFactory;
  import org.jpos.space.*;
  import org.jpos.util.*;
  
  import java.io.Serializable;
  import java.util.*;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.jpos.iso.ISOUtil;
  
  @SuppressWarnings("unchecked unused")
  public class TransactionManager 
      extends QBeanSupport 
      implements Runnable, TransactionConstants, TransactionManagerMBean
  {
      public static final String  HEAD       = "$HEAD";
      public static final String  TAIL       = "$TAIL";
      public static final String  CONTEXT    = "$CONTEXT.";
      public static final String  STATE      = "$STATE.";
      public static final String  GROUPS     = "$GROUPS.";
      public static final String  TAILLOCK   = "$TAILLOCK";
      public static final String  RETRY_QUEUE = "$RETRY_QUEUE";
      public static final Integer PREPARING  = 0;
      public static final Integer COMMITTING = 1;
      public static final Integer DONE       = 2;
      public static final String  DEFAULT_GROUP = "";
      public static final long    MAX_PARTICIPANTS = 1000;  // loop prevention
      public static final long    MAX_WAIT = 15000L;
      public static final long    TIMER_PURGE_INTERVAL = 1000L;
      protected Map<String,List<TransactionParticipant>> groups;
      private static final ThreadLocal<Serializable> tlContext = new ThreadLocal<Serializable>();
      private static final ThreadLocal<Long> tlId = new ThreadLocal<Long>();
  
      Space sp;
      Space psp;
      Space isp; // input space
      String queue;
      String tailLock;
      List<Thread> threads;
      final List<TransactionStatusListener> statusListeners = new ArrayList<TransactionStatusListener>();
      boolean hasStatusListeners;
      boolean debug;
      boolean profiler;
      boolean doRecover;
      boolean callSelectorOnAbort;
      int sessions;
      int maxSessions;
      int threshold;
      int maxActiveSessions;
      AtomicInteger activeSessions = new AtomicInteger();
      long head, tail;
      long retryInterval = 5000L;
      long retryTimeout  = 60000L;
      long pauseTimeout  = 0L;
      Runnable retryTask = null;
      TPS tps;
      final Timer timer = DefaultTimer.getTimer();
  
      @Override
      public void initService () throws ConfigurationException {
          queue = cfg.get ("queue", null);
          if (queue == null)
              throw new ConfigurationException ("queue property not specified");
          sp   = SpaceFactory.getSpace (cfg.get ("space"));
          isp  = SpaceFactory.getSpace (cfg.get ("input-space", cfg.get ("space")));
          psp  = SpaceFactory.getSpace (cfg.get ("persistent-space", this.toString()));
          tail = initCounter (TAIL, cfg.getLong ("initial-tail", 1));
          head = Math.max (initCounter (HEAD, tail), tail);
          initTailLock ();
  
          groups = new HashMap<String,List<TransactionParticipant>>();
          initParticipants (getPersist());
          initStatusListeners (getPersist());
      }
  
      @Override
      public void startService () throws Exception {
          NameRegistrar.register(getName(), this);
          recover();
          threads = Collections.synchronizedList(new ArrayList(maxSessions));
          if (tps != null)
              tps.stop();
          tps = new TPS (cfg.getBoolean ("auto-update-tps", true));
          for (int i=0; i<sessions; i++) {
              new Thread(this).start();
          }
          if (psp.rdp (RETRY_QUEUE) != null)
              checkRetryTask();
      }
  
      @Override
      public void stopService () throws Exception {
          NameRegistrar.unregister(getName());
  
          Thread[] tt = threads.toArray(new Thread[threads.size()]);
          for (int i=0; i < tt.length; i++) {
              isp.out(queue, Boolean.FALSE, 60 * 1000);
          }
          for (Thread thread : tt) {
              try {
                  thread.join (60*1000);
                  threads.remove(thread);
              } catch (InterruptedException e) {
                  getLog().warn ("Session " + thread.getName() +" does not respond - attempting to interrupt");
                  thread.interrupt();
              }
          }
          tps.stop();
      }
      public void queue (Serializable context) {
          isp.out(queue, context);
      }
      public void push (Serializable context) {
          isp.push(queue, context);
      }
      @SuppressWarnings("unused")
      public String getQueueName() {
          return queue;
      }
      public Space getSpace() {
          return sp;
      }
      public Space getInputSpace() {
          return isp;
      }
      public Space getPersistentSpace() {
          return psp;
      }
  
      @Override
      public void run () {
          long id = 0;
          int session = 0; // FIXME
          List<TransactionParticipant> members = null;
          Iterator<TransactionParticipant> iter = null;
          PausedTransaction pt;
          boolean abort = false;
          LogEvent evt = null;
          Profiler prof = null;
          long startTime = 0L;
          boolean paused;
          Thread thread = Thread.currentThread();
          if (threads.size() < maxSessions) {
              threads.add(thread);
              session = threads.indexOf(thread);
              activeSessions.incrementAndGet();
          } else {
              getLog().warn ("Max sessions reached, new session not created");
              return;
          }
          getLog().info ("start " + thread);
          while (running()) {
              Serializable context = null;
              prof = null;
              paused = false;
              thread.setName (getName() + "-" + session + ":idle");
              try {
                  if (hasStatusListeners)
                      notifyStatusListeners (session, TransactionStatusEvent.State.READY, id, "", null);
  
                  Object obj = isp.in (queue, MAX_WAIT);
                  if (obj == Boolean.FALSE)
                      continue;   // stopService ``hack''
  
                  if (obj == null) {
                      if (session > sessions && getActiveSessions() > sessions)
                          break; // we are an extra session, exit
                      else
                          continue;
                  }
                  if (session < sessions && // only initial sessions create extra sessions
                      maxSessions > sessions &&
                      getActiveSessions() < maxSessions &&
                      getOutstandingTransactions() > threshold)
                  {
                          new Thread(this).start();
                  }
                  if (!(obj instanceof Serializable)) {
                      getLog().error (
                          "non serializable '" + obj.getClass().getName()
                         + "' on queue '" + queue + "'"
                      );
                      continue;
                  }
                  context = (Serializable) obj;
                  if (obj instanceof Pausable) {
                      Pausable pausable = (Pausable) obj;
  