X-Git-Url: https://git.pterodactylus.net/?p=sonitus.git;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Fsonitus%2Fdata%2FPipeline.java;h=ae4158df1614cc1bcf7b60d5871cf8805c7adeee;hp=a38d6fe62d1e2f7ae3a73066f613740b8e89df38;hb=ef71e567bf3eea60be94a5c56aa8f9bc423dd0f0;hpb=6ff6dcb1223b2c948b2456b5d4e63c9a8ee0971d diff --git a/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java b/src/main/java/net/pterodactylus/sonitus/data/Pipeline.java index a38d6fe..ae4158d 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,7 +167,12 @@ public class Pipeline implements Iterable { } } for (Connection connection : connections) { - logger.info(String.format("Starting Connection from %s to %s.", connection.source, connection.sinks)); + logger.info(String.format("Starting Connection from %s to %s.", connection.source.name(), FluentIterable.from(connection.sinks).transform(new Function() { + @Override + public String apply(Sink sink) { + return sink.name(); + } + }))); new Thread(connection).start(); } } @@ -149,8 +192,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 +330,9 @@ public class Pipeline implements Iterable { /** The executor service. */ private final ExecutorService executorService; + /** The number of copied bytes. */ + private long counter; + /** * Creates a new connection. * @@ -279,6 +352,20 @@ public class Pipeline implements Iterable { } // + // ACCESSORS + // + + /** + * Returns the number of bytes that this connection has received from its + * source during its lifetime. + * + * @return The number of processed input bytes + */ + public long counter() { + return counter; + } + + // // ACTIONS // @@ -296,8 +383,6 @@ public class Pipeline implements Iterable { Metadata firstMetadata = null; 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)); @@ -314,9 +399,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); @@ -333,6 +415,7 @@ public class Pipeline implements Iterable { for (Future future : futures) { future.get(); } + counter += buffer.length; } catch (IOException e) { /* TODO */ e.printStackTrace(); @@ -351,4 +434,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); + } + + } + }