protected void processConnection(StreamConnection connection) {
		LOGGER.entry(connection);

		if (stopRequested) {
			LOGGER.exit();
			return;
		}

		DataInputStream dis;
		try {
			dis = connection.openDataInputStream();
		} catch (IOException e) {
			LOGGER.catching(e);
			LOGGER.exit();
			return;
		}

		boolean connectionOpen = true;

		try {
			while (!stopRequested && connectionOpen) {

				// Get ready to publish the message
				long sequence = disruptor.next();
				Message event = disruptor.get(sequence);
				event.reset();

				// Read data from the socket, place into the event
				// This will publish the message when "-1" is returned as well.
				int len = dis.read(event.getBuffer());
				event.setLength(len);

				// Publish the event
				disruptor.publish(sequence);

				if (len == -1) {
					connectionOpen = false;
				}
			}

		} catch (IOException e) {
			LOGGER.catching(e);
		} finally {
			try {
				connection.close();
			} catch (IOException e) {
				LOGGER.catching(e);
			}
		}

		LOGGER.exit();
	}