src/main/java/com/findupon/cluster/master/MasterNodeAgency.java
/**
	 * If the node is waiting, transition its state and move to the work pool.
	 * If the node is already working on another source, update its description and carry on.
	 */
If the node is waiting, transition its state and move to the work pool.
If the node is already working on another source, update its description and carry on.
If the node is waiting transition its state and move to the work pool . If the node is already working on another source update its description and carry on .
void workTransition(WorkerNode node, ClusterTransmission transmission) {
		synchronized(mutex) {
			boolean alreadyWorking = workingNodes.contains(node);
			if(alreadyWorking) {
				logger.trace("[Node {}] - Already working, updating work description for work transition...", node.getId());
			} else {
				logger.trace("[Node {}] - Transitioning to working...", node.getId());
			}
			if(MessageType.WORK_ORDER.equals(transmission.getMessage().getType())) {
				com.findupon.commons.entity.datasource.DataSource ds = transmission.getDataSource();
				if(ds == null || ds.getAssetType() == null || StringUtils.isEmpty(ds.getUrl())) {
					logger.error("[Node {}] - Invalid datasource on transmission, transition to working fails! [{}]",
							node.getId(), ds, new UnsupportedOperationException());
					return;
				} else {
					if(alreadyWorking && StringUtils.isNotEmpty(node.getWorkDescription())) {
						List<String> workingSources = new ArrayList<>(Arrays.asList(node.getWorkDescription().split(" ~ ")));
						workingSources.add(ds.getUrl());
						node.setWorkDescription(workingSources.stream().collect(Collectors.joining(" ~ ")));
					} else {
						node.setWorkDescription(ds.getUrl());
					}
				}
			}
			if(!alreadyWorking) {
				node.setIsWorking(true);
				if(waitingNodes.remove(node)) {
					workingNodes.add(node);
					workerNodeRepo.saveAndFlush(node);
					logger.debug("[Node {}] - Transitioned to working", node.getId());
				} else {
					logger.warn("[Node {}] - Transition to working failed! Not in waiting nodes", node.getId());
				}
			} else {
				workerNodeRepo.saveAndFlush(node);
			}
		}
		logger.trace("[Node {}] - Working transition lock released", node.getId());
	}
void workTransition ( WorkerNode node , ClusterTransmission transmission ) { synchronized ( mutex ) { boolean alreadyWorking = workingNodes . contains ( node ) ; if ( alreadyWorking ) { logger . trace ( "[Node {}] - Already working, updating work description for work transition..." , node . getId ( ) ) ; } else { logger . trace ( "[Node {}] - Transitioning to working..." , node . getId ( ) ) ; } if ( MessageType . WORK_ORDER . equals ( transmission . getMessage ( ) . getType ( ) ) ) { com . findupon . commons . entity . datasource . DataSource ds = transmission . getDataSource ( ) ; if ( ds == null || ds . getAssetType ( ) == null || StringUtils . isEmpty ( ds . getUrl ( ) ) ) { logger . error ( "[Node {}] - Invalid datasource on transmission, transition to working fails! [{}]" , node . getId ( ) , ds , new UnsupportedOperationException ( ) ) ; return ; } else { if ( alreadyWorking && StringUtils . isNotEmpty ( node . getWorkDescription ( ) ) ) { List < String > workingSources = new ArrayList < > ( Arrays . asList ( node . getWorkDescription ( ) . split ( " ~ " ) ) ) ; workingSources . add ( ds . getUrl ( ) ) ; node . setWorkDescription ( workingSources . stream ( ) . collect ( Collectors . joining ( " ~ " ) ) ) ; } else { node . setWorkDescription ( ds . getUrl ( ) ) ; } } } if ( ! alreadyWorking ) { node . setIsWorking ( true ) ; if ( waitingNodes . remove ( node ) ) { workingNodes . add ( node ) ; workerNodeRepo . saveAndFlush ( node ) ; logger . debug ( "[Node {}] - Transitioned to working" , node . getId ( ) ) ; } else { logger . warn ( "[Node {}] - Transition to working failed! Not in waiting nodes" , node . getId ( ) ) ; } } else { workerNodeRepo . saveAndFlush ( node ) ; } } logger . trace ( "[Node {}] - Working transition lock released" , node . getId ( ) ) ; }
If the node is waiting, transition its state and move to the work pool.
If the node is waiting transition its state and move to the work pool .