docs: @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);
  				}
  			}
  		}
  	}