Commit 6a8700a3 authored by Florian Fittkau's avatar Florian Fittkau

WiP

parent 195f5c7d
......@@ -4,6 +4,8 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.ServerSocketChannel;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
......@@ -26,6 +28,8 @@ public final class TCPReader implements IPeriodicTimeSignalReceiver {
private final Queue<IRecord> periodicSignalQueue;
private final ExecutorService threadPool;
public TCPReader(final int listeningPort, final PipesMerger<IRecord> traceReconstructionMerger) {
this.listeningPort = listeningPort;
......@@ -33,6 +37,8 @@ public final class TCPReader implements IPeriodicTimeSignalReceiver {
new TimeSignalReader(TimeUnit.SECONDS.toMillis(1), this).start();
periodicSignalQueue = merger.registerProducer();
threadPool = Executors.newCachedThreadPool();
}
@Override
......@@ -50,7 +56,7 @@ public final class TCPReader implements IPeriodicTimeSignalReceiver {
try {
open();
while (active) {
new TCPReaderOneClient(serversocket.accept(), merger).start();
threadPool.execute(new TCPReaderOneClient(serversocket.accept(), merger));
}
} catch (final IOException ex) {
LOG.info("Error in read() " + ex.getMessage());
......@@ -72,8 +78,10 @@ public final class TCPReader implements IPeriodicTimeSignalReceiver {
public final void terminate(final boolean error) {
LOG.info("Shutdown of TCPReader requested.");
active = false;
// for (final TCPReaderOneClient thread : threads) {
// thread.terminate();
// }
try {
threadPool.awaitTermination(10, TimeUnit.SECONDS);
} catch (final InterruptedException e) {
}
threadPool.shutdown();
}
}
......@@ -34,7 +34,7 @@ import explorviz.live_trace_processing.record.misc.StringRegistryRecord;
import explorviz.live_trace_processing.record.misc.SystemMonitoringRecord;
import explorviz.live_trace_processing.record.trace.HostApplicationMetaDataRecord;
class TCPReaderOneClient extends Thread {
class TCPReaderOneClient implements Runnable {
private static final Logger LOG = LoggerFactory.getLogger(TCPReaderOneClient.class);
......@@ -895,14 +895,12 @@ class TCPReaderOneClient extends Thread {
return;
}
System.out.println("waitingForStringMessages: " + waitingForStringMessages.size());
final List<byte[]> localWaitingList = new ArrayList<byte[]>(waitingForStringMessages);
waitingForStringMessages.clear();
for (final byte[] waitingMessage : localWaitingList) {
final ByteBuffer buffer = ByteBuffer.wrap(waitingMessage);
final byte waitingMessageClazzId = buffer.get();
System.out.println("waitingForStringMessages clazzId: " + waitingMessageClazzId);
switch (waitingMessageClazzId) {
case HostApplicationMetaDataRecord.CLAZZ_ID:
readInHostApplicationMetaData(buffer);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment