src/main/java/com/findupon/cluster/master/MasterWorkDelegate.java
/**
	 * Send requests to nodes based on their allowed work type.
	 * If a node is already working on a listing data source, it is not allowed another one.
	 * If the work queue polls a type that is not allowed, put it back on the front of the queue after a request is sent or no valid type is found.
	 */
Send requests to nodes based on their allowed work type.
If a node is already working on a listing data source, it is not allowed another one.
If the work queue polls a type that is not allowed, put it back on the front of the queue after a request is sent or no valid type is found.
Send requests to nodes based on their allowed work type . If a node is already working on a listing data source it is not allowed another one . If the work queue polls a type that is not allowed put it back on the front of the queue after a request is sent or no valid type is found .
private void sendWorkRequests(Map<WorkerNode, AllowedWorkType> readyNodes) {
		for(Map.Entry<WorkerNode, AllowedWorkType> entry : readyNodes.entrySet()) {
			if(workQueue.isEmpty()) {
				logger.debug("[MasterWorkDelegate] - No requests to send, work queue is empty");
				break;
			}
			if(!run.get()) {
				break;
			}
			List<ClusterTransmission> transmissionsToPutBack = new ArrayList<>();
			AllowedWorkType allowedWorkType = entry.getValue();
			WorkerNode node = entry.getKey();
			ClusterTransmission nextWorkOrder;
			while(true) {
				if((nextWorkOrder = workQueue.poll()) == null || !run.get()) {
					transmissionsToPutBack.forEach(workQueue::offerFirst);
					break;
				} else if(AllowedWorkType.AGNOSTIC.equals(allowedWorkType)
						|| (AllowedWorkType.GENERIC.equals(allowedWorkType) && nextWorkOrder.getDataSource().isGeneric())
						|| (AllowedWorkType.LISTING.equals(allowedWorkType) && !nextWorkOrder.getDataSource().isGeneric())) {
					boolean clearToSend = true;
					if(!nextWorkOrder.getDataSource().isGeneric()) {
						for(SentRequest existingRequest : getSentWorkRequestsSnapshotForNode(node)) {
							if(nextWorkOrder.getDataSource().equals(existingRequest.getTransmission().getDataSource())) {
								transmissionsToPutBack.add(nextWorkOrder);
								clearToSend = false;
								break;
							}
						}
					}
					if(clearToSend) {
						transmissionsToPutBack.forEach(workQueue::offerFirst);
						break;
					}
				} else {
					transmissionsToPutBack.add(nextWorkOrder);
				}
			}
			if(nextWorkOrder != null && run.get()) {
				nextWorkOrder.setNodeId(node.getId());
				try {
					masterNodeAgency.sendTransmission(node, nextWorkOrder);
				} catch(Exception e) {
					logger.warn("[Node {}] - Did not receive work order! Adding back to the work queue. Cause [{}]",
							node.getId(), ExceptionUtils.getRootCauseMessage(e));
					nextWorkOrder.setNodeId(null);
					workQueue.offerFirst(nextWorkOrder);
				}
			}
		}
	}
private void sendWorkRequests ( Map < WorkerNode , AllowedWorkType > readyNodes ) { for ( Map . Entry < WorkerNode , AllowedWorkType > entry : readyNodes . entrySet ( ) ) { if ( workQueue . isEmpty ( ) ) { logger . debug ( "[MasterWorkDelegate] - No requests to send, work queue is empty" ) ; break ; } if ( ! run . get ( ) ) { break ; } List < ClusterTransmission > transmissionsToPutBack = new ArrayList < > ( ) ; AllowedWorkType allowedWorkType = entry . getValue ( ) ; WorkerNode node = entry . getKey ( ) ; ClusterTransmission nextWorkOrder ; while ( true ) { if ( ( nextWorkOrder = workQueue . poll ( ) ) == null || ! run . get ( ) ) { transmissionsToPutBack . forEach ( workQueue :: offerFirst ) ; break ; } else if ( AllowedWorkType . AGNOSTIC . equals ( allowedWorkType ) || ( AllowedWorkType . GENERIC . equals ( allowedWorkType ) && nextWorkOrder . getDataSource ( ) . isGeneric ( ) ) || ( AllowedWorkType . LISTING . equals ( allowedWorkType ) && ! nextWorkOrder . getDataSource ( ) . isGeneric ( ) ) ) { boolean clearToSend = true ; if ( ! nextWorkOrder . getDataSource ( ) . isGeneric ( ) ) { for ( SentRequest existingRequest : getSentWorkRequestsSnapshotForNode ( node ) ) { if ( nextWorkOrder . getDataSource ( ) . equals ( existingRequest . getTransmission ( ) . getDataSource ( ) ) ) { transmissionsToPutBack . add ( nextWorkOrder ) ; clearToSend = false ; break ; } } } if ( clearToSend ) { transmissionsToPutBack . forEach ( workQueue :: offerFirst ) ; break ; } } else { transmissionsToPutBack . add ( nextWorkOrder ) ; } } if ( nextWorkOrder != null && run . get ( ) ) { nextWorkOrder . setNodeId ( node . getId ( ) ) ; try { masterNodeAgency . sendTransmission ( node , nextWorkOrder ) ; } catch ( Exception e ) { logger . warn ( "[Node {}] - Did not receive work order! Adding back to the work queue. Cause [{}]" , node . getId ( ) , ExceptionUtils . getRootCauseMessage ( e ) ) ; nextWorkOrder . setNodeId ( null ) ; workQueue . offerFirst ( nextWorkOrder ) ; } } } }
Send requests to nodes based on their allowed work type.
Send requests to nodes based on their allowed work type .
// go through the work queue looking for an order the node can accept // no more work requests, put any polled back at the front and continue to the next node // make sure the node is not working on the same listing data source // poll sent requests again as its state may have changed // found an allowed work request, put any polled back at the front and send the request to the node // work request not allowed, save to put back later