@Override
	public void invoke(IN value) throws Exception {
		byte[] msg = schema.serialize(value);

		try {
			outputStream.write(msg);
			if (autoFlush) {
				outputStream.flush();
			}
		}
		catch (IOException e) {
			// if no re-tries are enable, fail immediately
			if (maxNumRetries == 0) {
				throw new IOException("Failed to send message '" + value + "' to socket server at "
						+ hostName + ":" + port + ". Connection re-tries are not enabled.", e);
			}

			LOG.error("Failed to send message '" + value + "' to socket server at " + hostName + ":" + port +
					". Trying to reconnect..." , e);

			// do the retries in locked scope, to guard against concurrent close() calls
			// note that the first re-try comes immediately, without a wait!

			synchronized (lock) {
				IOException lastException = null;
				retries = 0;

				while (isRunning && (maxNumRetries < 0 || retries < maxNumRetries)) {

					// first, clean up the old resources
					try {
						if (outputStream != null) {
							outputStream.close();
						}
					}
					catch (IOException ee) {
						LOG.error("Could not close output stream from failed write attempt", ee);
					}
					try {
						if (client != null) {
							client.close();
						}
					}
					catch (IOException ee) {
						LOG.error("Could not close socket from failed write attempt", ee);
					}

					// try again
					retries++;

					try {
						// initialize a new connection
						createConnection();

						// re-try the write
						outputStream.write(msg);

						// success!
						return;
					}
					catch (IOException ee) {
						lastException = ee;
						LOG.error("Re-connect to socket server and send message failed. Retry time(s): " + retries, ee);
					}

					// wait before re-attempting to connect
					lock.wait(CONNECTION_RETRY_DELAY);
				}

				// throw an exception if the task is still running, otherwise simply leave the method
				if (isRunning) {
					throw new IOException("Failed to send message '" + value + "' to socket server at "
							+ hostName + ":" + port + ". Failed after " + retries + " retries.", lastException);
				}
			}
		}
	}
@ Override public void invoke ( IN value ) throws Exception { byte [ ] msg = schema . serialize ( value ) ; try { outputStream . write ( msg ) ; if ( autoFlush ) { outputStream . flush ( ) ; } } catch ( IOException e ) { if ( maxNumRetries == 0 ) { throw new IOException ( "Failed to send message '" + value + "' to socket server at " + hostName + ":" + port + ". Connection re-tries are not enabled." , e ) ; } LOG . error ( "Failed to send message '" + value + "' to socket server at " + hostName + ":" + port + ". Trying to reconnect..." , e ) ; synchronized ( lock ) { IOException lastException = null ; retries = 0 ; while ( isRunning && ( maxNumRetries < 0 || retries < maxNumRetries ) ) { try { if ( outputStream != null ) { outputStream . close ( ) ; } } catch ( IOException ee ) { LOG . error ( "Could not close output stream from failed write attempt" , ee ) ; } try { if ( client != null ) { client . close ( ) ; } } catch ( IOException ee ) { LOG . error ( "Could not close socket from failed write attempt" , ee ) ; } retries ++ ; try { createConnection ( ) ; outputStream . write ( msg ) ; return ; } catch ( IOException ee ) { lastException = ee ; LOG . error ( "Re-connect to socket server and send message failed. Retry time(s): " + retries , ee ) ; } lock . wait ( CONNECTION_RETRY_DELAY ) ; } if ( isRunning ) { throw new IOException ( "Failed to send message '" + value + "' to socket server at " + hostName + ":" + port + ". Failed after " + retries + " retries." , lastException ) ; } } } }
Called when new data arrives to the sink, and forwards it to Socket.

@param value The value to write to the socket.