protected boolean internalRun()

in modules/core/src/main/java/org/apache/sandesha2/polling/PollingManager.java [74:138]
53 lines of code
20 McCabe index (conditional complexity)

	protected boolean internalRun() {
		if(log.isDebugEnabled()) log.debug("Enter: PollingManager::internalRun");
		Transaction t = null;
		try {
			// If we have request scheduled, handle them first, and then pick
			// pick a sequence using a round-robin approach. Scheduled polls
			// bypass the normal polling checks, to make sure that they happen
			boolean forcePoll = false;
			SequenceEntry entry = null;
			synchronized (this) {
				if(!scheduledPollingRequests.isEmpty()) {
					entry = (SequenceEntry) scheduledPollingRequests.removeFirst();
					forcePoll = true;
				}
			}
			if(entry == null) {
				ArrayList<SequenceEntry> allSequencesList = getSequences();
				int size = allSequencesList.size();
				if(log.isDebugEnabled()) log.debug("Choosing one from " + size + " sequences");
				if(nextIndex >= size) {
					nextIndex = 0;
					// We just looped over the set of sequences, so sleep before we try
					// polling them again.
					if (log.isDebugEnabled()) log.debug("Exit: PollingManager::internalRun, looped over all sequences, sleeping");
					return true;
				}
	
				entry = (SequenceEntry) allSequencesList.get(nextIndex++);
				
				long now = System.currentTimeMillis();
				Long time = (Long) pollTimes.get(entry.getSequenceId());
				if(time != null && (now - time.longValue()) < POLLING_MANAGER_WAIT_TIME) {
					// We have polled for this sequence too recently, so skip over this one
					if (log.isDebugEnabled()) log.debug("Exit: PollingManager::internalRun, skipping sequence, not sleeping");
					return false;
				}
				pollTimes.put(entry.getSequenceId(), new Long(now));
				
			}
			if(log.isDebugEnabled()) log.debug("Chose sequence " + entry.getSequenceId());

			t = storageManager.getTransaction();
			if(entry.isRmSource()) {
				pollRMSSide(entry, forcePoll);
			} else {
				pollRMDSide(entry, forcePoll);
			}
			if(t != null) t.commit();
			t = null;

		} catch (Exception e) {
			if(log.isDebugEnabled()) log.debug("Exception", e);
		} finally {
			if(t != null && t.isActive()) {
				try {
					t.rollback();
				} catch(Exception e2) {
					if(log.isDebugEnabled()) log.debug("Exception during rollback", e2);
				}
			}
		}
		
		if(log.isDebugEnabled()) log.debug("Exit: PollingManager::internalRun, not sleeping");
		return false;
	}