src/main/java/com/findupon/cluster/master/MasterNodeAgency.java
// TODO: make this concurrent by implementing a send pool similar to the response pool
make this concurrent by implementing a send pool similar to the response pool
make this concurrent by implementing a send pool similar to the response pool
void sendTransmission(WorkerNode node, ClusterTransmission transmission) throws IOException {
		SentRequest sentRequest = new SentRequest(node, new Date(), transmission);
		if(!MessageType.NO_RESPONSE.equals(transmission.getMessage().getType())) {
			sentRequests.add(sentRequest);
		}
		for(int attempt = 1; attempt <= SOCKET_CONNECT_MAX_ATTEMPTS; attempt++) {
			boolean success = true;
			try {
				internalSendTransmission(sentRequest);
			} catch(IOException e) {
				if(!master.getRunning()) {
					return;
				}
				success = false;
				if(attempt == SOCKET_CONNECT_MAX_ATTEMPTS) {
					logger.error("Error sending transmission [{}] to node. Final attempt [{}/{}]. Removing sent request. Node [{} - {}:{}]",
							transmission.getMessage().getDirective(), attempt, SOCKET_CONNECT_MAX_ATTEMPTS, node.getId(), node.getAddress(), node.getPort(), e);
					sentRequests.remove(sentRequest);
					throw e;
				} else {
					logger.warn("Error sending transmission [{}] to node. Attempt [{}/{}]. Node [{} - {}:{}]",
							transmission.getMessage().getDirective(), attempt, SOCKET_CONNECT_MAX_ATTEMPTS, node.getId(), node.getAddress(), node.getPort(), e);
				}
			}
			if(success) {
				break;
			}
		}
	}
void sendTransmission ( WorkerNode node , ClusterTransmission transmission ) throws IOException { SentRequest sentRequest = new SentRequest ( node , new Date ( ) , transmission ) ; if ( ! MessageType . NO_RESPONSE . equals ( transmission . getMessage ( ) . getType ( ) ) ) { sentRequests . add ( sentRequest ) ; } for ( int attempt = 1 ; attempt <= SOCKET_CONNECT_MAX_ATTEMPTS ; attempt ++ ) { boolean success = true ; try { internalSendTransmission ( sentRequest ) ; } catch ( IOException e ) { if ( ! master . getRunning ( ) ) { return ; } success = false ; if ( attempt == SOCKET_CONNECT_MAX_ATTEMPTS ) { logger . error ( "Error sending transmission [{}] to node. Final attempt [{}/{}]. Removing sent request. Node [{} - {}:{}]" , transmission . getMessage ( ) . getDirective ( ) , attempt , SOCKET_CONNECT_MAX_ATTEMPTS , node . getId ( ) , node . getAddress ( ) , node . getPort ( ) , e ) ; sentRequests . remove ( sentRequest ) ; throw e ; } else { logger . warn ( "Error sending transmission [{}] to node. Attempt [{}/{}]. Node [{} - {}:{}]" , transmission . getMessage ( ) . getDirective ( ) , attempt , SOCKET_CONNECT_MAX_ATTEMPTS , node . getId ( ) , node . getAddress ( ) , node . getPort ( ) , e ) ; } } if ( success ) { break ; } } }
TODO: make this concurrent by implementing a send pool similar to the response pool
TODO : make this concurrent by implementing a send pool similar to the response pool