envoy/src/main/java/envoy/client/net/Receiver.java

package envoy.client.net;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.net.SocketException;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
import java.util.logging.Level;
import java.util.logging.Logger;

import envoy.client.util.EnvoyLog;
import envoy.util.SerializationUtils;

/**
 * Project: <strong>envoy-client</strong><br>
 * File: <strong>Receiver.java</strong><br>
 * Created: <strong>30.12.2019</strong><br>
 *
 * @author Kai S. K. Engelbart
 * @since Envoy v0.3-alpha
 */
public class Receiver implements Runnable {

	private final InputStream					in;
	private final Map<Class<?>, Consumer<?>>	processors	= new HashMap<>();

	private static final Logger logger = EnvoyLog.getLogger(Receiver.class.getSimpleName());

	/**
	 * Creates an instance of {@link Receiver}.
	 *
	 * @param in the {@link InputStream} to parse objects from
	 */
	public Receiver(InputStream in) { this.in = in; }

	@Override
	public void run() {

		try {
			while (true) {
				// Read object length
				byte[] lenBytes = new byte[4];
				in.read(lenBytes);
				int len = SerializationUtils.bytesToInt(lenBytes, 0);

				// Read object into byte array
				byte[] objBytes = new byte[len];
				in.read(objBytes);

				try (ObjectInputStream oin = new ObjectInputStream(new ByteArrayInputStream(objBytes))) {
					Object obj = oin.readObject();
					logger.info("Received object " + obj);

					// Get appropriate processor
					@SuppressWarnings("rawtypes")
					Consumer processor = processors.get(obj.getClass());
					if (processor == null)
						logger.severe(String.format("The received object has the class %s for which no processor is defined.", obj.getClass()));
					else processor.accept(obj);
				}
			}
		} catch (SocketException e) {
			logger.info("Connection probably closed by client. Exiting receiver thread...");
		} catch (Exception e) {
			logger.log(Level.SEVERE, "Error on receiver thread", e);
			e.printStackTrace();
		}
	}

	/**
	 * Adds an object processor to this {@link Receiver}. It will be called once an
	 * object of the accepted class has been received.
	 *
	 * @param processorClass the object class accepted by the processor
	 * @param processor      the object processor
	 */
	public <T> void registerProcessor(Class<T> processorClass, Consumer<T> processor) { processors.put(processorClass, processor); }

	/**
	 * Removes all object processors registered at this {@link Receiver}.
	 */
	public void removeAllProcessors() { processors.clear(); }
}