X-Git-Url: https://git.pterodactylus.net/?p=sonitus.git;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Fsonitus%2Fdata%2FPipeline.java;h=063a0bc52e75023cee02dd4078c0e7b88643bd37;hp=4409151fba9cfeeb05a0c4451fc865a689203141;hb=44e662f7a3918b46c3abe6646c1e167a112749c1;hpb=57737deafd02c03fcea5c3a3465d910338558f29 diff --git a/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java b/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java index 4409151..063a0bc 100644 --- a/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java +++ b/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java @@ -30,11 +30,12 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.logging.Logger; import com.google.common.base.Function; +import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.FluentIterable; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMultimap; -import com.google.common.collect.ImmutableSet; import com.google.common.collect.Lists; import com.google.common.collect.Multimap; import com.google.common.util.concurrent.MoreExecutors; @@ -45,7 +46,7 @@ import com.google.common.util.concurrent.MoreExecutors; * * @author David ‘Bombe’ Roden */ -public class Pipeline implements Iterable { +public class Pipeline implements Iterable { /** The logger. */ private static final Logger logger = Logger.getLogger(Pipeline.class.getName()); @@ -70,6 +71,21 @@ public class Pipeline implements Iterable { private Pipeline(Source source, Multimap sinks) { this.source = Preconditions.checkNotNull(source, "source must not be null"); this.sinks = Preconditions.checkNotNull(sinks, "sinks must not be null"); + for (ControlledComponent component : Lists.reverse(components())) { + logger.finest(String.format("Adding Listener to %s.", component.name())); + component.addMetadataListener(new MetadataListener() { + @Override + public void metadataUpdated(ControlledComponent component, Metadata metadata) { + if (!(component instanceof Source)) { + return; + } + for (ControlledComponent controlledComponent : sinks((Source) component)) { + logger.fine(String.format("Updating Metadata from %s to %s as %s.", component.name(), controlledComponent.name(), metadata)); + controlledComponent.metadataUpdated(metadata); + } + } + }); + } } // @@ -98,6 +114,27 @@ public class Pipeline implements Iterable { return sinks.get(source); } + /** + * Returns the traffic counters of the given controlled component. + * + * @param controlledComponent + * The controlled component to get the traffic counters for + * @return The traffic counters for the given controlled component + */ + public TrafficCounter trafficCounter(ControlledComponent controlledComponent) { + long input = -1; + long output = -1; + for (Connection connection : connections) { + /* the connection where the source matches knows the output. */ + if (connection.source.equals(controlledComponent)) { + output = connection.counter(); + } else if (connection.sinks.contains(controlledComponent)) { + input = connection.counter(); + } + } + return new TrafficCounter(input, output); + } + // // ACTIONS // @@ -122,6 +159,7 @@ public class Pipeline implements Iterable { Collection sinks = this.sinks.get(source); connections.add(new Connection(source, sinks)); for (Sink sink : sinks) { + logger.info(String.format("Opening %s with %s...", sink.name(), source.metadata())); sink.open(source.metadata()); if (sink instanceof Filter) { sources.add((Source) sink); @@ -129,8 +167,15 @@ public class Pipeline implements Iterable { } } for (Connection connection : connections) { - logger.info(String.format("Starting Connection from %s to %s.", connection.source, connection.sinks)); - new Thread(connection).start(); + String threadName = String.format("%s → %s.", connection.source.name(), FluentIterable.from(connection.sinks).transform(new Function() { + + @Override + public String apply(Sink sink) { + return sink.name(); + } + })); + logger.info(String.format("Starting Thread: %s", threadName)); + new Thread(connection, threadName).start(); } } @@ -149,8 +194,35 @@ public class Pipeline implements Iterable { // @Override - public Iterator iterator() { - return ImmutableSet.builder().add(source).addAll(sinks.values()).build().iterator(); + public Iterator iterator() { + return components().iterator(); + } + + // + // PRIVATE METHODS + // + + /** + * Returns all components of this pipeline, listed breadth-first, starting with + * the source. + * + * @return All components of this pipeline + */ + public List components() { + ImmutableList.Builder components = ImmutableList.builder(); + List currentComponents = Lists.newArrayList(); + components.add(source); + currentComponents.add(source); + while (!currentComponents.isEmpty()) { + Collection sinks = this.sinks((Source) currentComponents.remove(0)); + for (Sink sink : sinks) { + components.add(sink); + if (sink instanceof Source) { + currentComponents.add(sink); + } + } + } + return components.build(); } // @@ -260,6 +332,9 @@ public class Pipeline implements Iterable { /** The executor service. */ private final ExecutorService executorService; + /** The time the connection was started. */ + private long startTime; + /** The number of copied bytes. */ private long counter; @@ -274,7 +349,7 @@ public class Pipeline implements Iterable { public Connection(Source source, Collection sinks) { this.source = source; this.sinks = sinks; - if (sinks.size() == 1) { + if (sinks.size() < 2) { executorService = MoreExecutors.sameThreadExecutor(); } else { executorService = Executors.newCachedThreadPool(); @@ -286,6 +361,16 @@ public class Pipeline implements Iterable { // /** + * Returns the time this connection was started. + * + * @return The time this connection was started (in milliseconds since Jan 1, + * 1970 UTC) + */ + public long startTime() { + return startTime; + } + + /** * Returns the number of bytes that this connection has received from its * source during its lifetime. * @@ -310,11 +395,9 @@ public class Pipeline implements Iterable { @Override public void run() { - Metadata firstMetadata = null; + startTime = System.currentTimeMillis(); while (!stopped.get()) { try { - final Metadata lastMetadata = firstMetadata; - final Metadata metadata = firstMetadata = source.metadata(); final byte[] buffer; try { logger.finest(String.format("Getting %d bytes from %s...", 4096, source)); @@ -331,9 +414,6 @@ public class Pipeline implements Iterable { @Override public Void call() throws Exception { - if (!metadata.equals(lastMetadata)) { - sink.metadataUpdated(metadata); - } try { logger.finest(String.format("Sending %d bytes to %s.", buffer.length, sink)); sink.process(buffer); @@ -369,4 +449,58 @@ public class Pipeline implements Iterable { } + /** + * Container for input and output counters. + * + * @author David ‘Bombe’ Roden + */ + public static class TrafficCounter { + + /** The number of input bytes. */ + private final long input; + + /** The number of output bytes. */ + private final long output; + + /** + * Creates a new traffic counter. + * + * @param input + * The number of input bytes (may be {@code -1} to signify non-available + * input) + * @param output + * The number of output bytes (may be {@code -1} to signify non-available + * output) + */ + public TrafficCounter(long input, long output) { + this.input = input; + this.output = output; + } + + // + // ACCESSORS + // + + /** + * Returns the number of input bytes. + * + * @return The number of input bytes, or {@link Optional#absent()} if the + * component can not receive input + */ + public Optional input() { + return (input == -1) ? Optional.absent() : Optional.of(input); + } + + /** + * Returns the number of output bytes. + * + * @return The number of output bytes, or {@link Optional#absent()} if the + * component can not send output + */ + public Optional output() { + return (output == -1) ? Optional.absent() : Optional.of(output); + } + + } + }