src/main/java/net/pterodactylus/sonitus/data/Pipeline.java

   1 /*
   2  * Sonitus - Pipeline.java - Copyright © 2013 David Roden
   3  *
   4  * This program is free software: you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License as published by
   6  * the Free Software Foundation, either version 3 of the License, or
   7  * (at your option) any later version.
   8  *
   9  * This program is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  * GNU General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License
  15  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16  */
  17
  18 package net.pterodactylus.sonitus.data;
  19
  20 import java.io.IOException;
  21 import java.util.Collection;
  22 import java.util.Iterator;
  23 import java.util.List;
  24 import java.util.concurrent.Callable;
  25 import java.util.concurrent.ExecutionException;
  26 import java.util.concurrent.ExecutorService;
  27 import java.util.concurrent.Executors;
  28 import java.util.concurrent.Future;
  29 import java.util.concurrent.atomic.AtomicBoolean;
  30 import java.util.logging.Logger;
  31
  32 import com.google.common.base.Function;
  33 import com.google.common.base.Optional;
  34 import com.google.common.base.Preconditions;
  35 import com.google.common.collect.ArrayListMultimap;
  36 import com.google.common.collect.FluentIterable;
  37 import com.google.common.collect.ImmutableList;
  38 import com.google.common.collect.ImmutableMultimap;
  39 import com.google.common.collect.Lists;
  40 import com.google.common.collect.Multimap;
  41 import com.google.common.util.concurrent.MoreExecutors;
  42
  43 /**
  44  * A pipeline is responsible for streaming audio data from a {@link Source} to
  45  * an arbitrary number of connected {@link Filter}s and {@link Sink}s.
  46  *
  47  * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
  48  */
  49 public class Pipeline implements Iterable<ControlledComponent> {
  50
  51         /** The logger. */
  52         private static final Logger logger = Logger.getLogger(Pipeline.class.getName());
  53
  54         /** The source of the audio stream. */
  55         private final Source source;
  56
  57         /** The sinks for each source. */
  58         private final Multimap<Source, Sink> sinks;
  59
  60         /** All started connections. */
  61         private final List<Connection> connections = Lists.newArrayList();
  62
  63         /**
  64          * Creates a new pipeline.
  65          *
  66          * @param source
  67          *              The source of the audio stream
  68          * @param sinks
  69          *              The sinks for each source
  70          */
  71         private Pipeline(Source source, Multimap<Source, Sink> sinks) {
  72                 this.source = Preconditions.checkNotNull(source, "source must not be null");
  73                 this.sinks = Preconditions.checkNotNull(sinks, "sinks must not be null");
  74                 for (ControlledComponent component : Lists.reverse(components())) {
  75                         logger.finest(String.format("Adding Listener to %s.", component));
  76                         component.addMetadataListener(new MetadataListener() {
  77                                 @Override
  78                                 public void metadataUpdated(ControlledComponent component, Metadata metadata) {
  79                                         if (!(component instanceof Source)) {
  80                                                 return;
  81                                         }
  82                                         for (ControlledComponent controlledComponent : sinks((Source) component)) {
  83                                                 logger.fine(String.format("Updating Metadata from %s to %s.", component, controlledComponent));
  84                                                 controlledComponent.metadataUpdated(metadata);
  85                                         }
  86                                 }
  87                         });
  88                 }
  89         }
  90
  91         //
  92         // ACCESSORS
  93         //
  94
  95         /**
  96          * Expose this pipeline’s source.
  97          *
  98          * @return This pipeline’s source
  99          */
 100         public Source source() {
 101                 return source;
 102         }
 103
 104         /**
 105          * Returns all {@link Sink}s (or {@link Filter}s, really) that are connected to
 106          * the given source.
 107          *
 108          * @param source
 109          *              The source to get the sinks for
 110          * @return The sinks connected to the given source, or an empty list if the
 111          *         source does not exist in this pipeline
 112          */
 113         public Collection<Sink> sinks(Source source) {
 114                 return sinks.get(source);
 115         }
 116
 117         /**
 118          * Returns the traffic counters of the given controlled component.
 119          *
 120          * @param controlledComponent
 121          *              The controlled component to get the traffic counters for
 122          * @return The traffic counters for the given controlled component
 123          */
 124         public TrafficCounter trafficCounter(ControlledComponent controlledComponent) {
 125                 long input = -1;
 126                 long output = -1;
 127                 for (Connection connection : connections) {
 128                         /* the connection where the source matches knows the output. */
 129                         if (connection.source.equals(controlledComponent)) {
 130                                 output = connection.counter();
 131                         } else if (connection.sinks.contains(controlledComponent)) {
 132                                 input = connection.counter();
 133                         }
 134                 }
 135                 return new TrafficCounter(input, output);
 136         }
 137
 138         //
 139         // ACTIONS
 140         //
 141
 142         /**
 143          * Starts the pipeline.
 144          *
 145          * @throws IOException
 146          *              if any of the sinks can not be opened
 147          * @throws IllegalStateException
 148          *              if the pipeline is already running
 149          */
 150         public void start() throws IOException, IllegalStateException {
 151                 if (!connections.isEmpty()) {
 152                         throw new IllegalStateException("Pipeline is already running!");
 153                 }
 154                 List<Source> sources = Lists.newArrayList();
 155                 sources.add(source);
 156                 /* collect all source->sink pairs. */
 157                 while (!sources.isEmpty()) {
 158                         Source source = sources.remove(0);
 159                         Collection<Sink> sinks = this.sinks.get(source);
 160                         connections.add(new Connection(source, sinks));
 161                         for (Sink sink : sinks) {
 162                                 sink.open(source.metadata());
 163                                 if (sink instanceof Filter) {
 164                                         sources.add((Source) sink);
 165                                 }
 166                         }
 167                 }
 168                 for (Connection connection : connections) {
 169                         logger.info(String.format("Starting Connection from %s to %s.", connection.source, connection.sinks));
 170                         new Thread(connection).start();
 171                 }
 172         }
 173
 174         public void stop() {
 175                 if (!connections.isEmpty()) {
 176                         /* pipeline is not running. */
 177                         return;
 178                 }
 179                 for (Connection connection : connections) {
 180                         connection.stop();
 181                 }
 182         }
 183
 184         //
 185         // ITERABLE METHODS
 186         //
 187
 188         @Override
 189         public Iterator<ControlledComponent> iterator() {
 190                 return components().iterator();
 191         }
 192
 193         //
 194         // PRIVATE METHODS
 195         //
 196
 197         /**
 198          * Returns all components of this pipeline, listed breadth-first, starting with
 199          * the source.
 200          *
 201          * @return All components of this pipeline
 202          */
 203         public List<ControlledComponent> components() {
 204                 ImmutableList.Builder<ControlledComponent> components = ImmutableList.builder();
 205                 List<ControlledComponent> currentComponents = Lists.newArrayList();
 206                 components.add(source);
 207                 currentComponents.add(source);
 208                 while (!currentComponents.isEmpty()) {
 209                         Collection<Sink> sinks = this.sinks((Source) currentComponents.remove(0));
 210                         for (Sink sink : sinks) {
 211                                 components.add(sink);
 212                                 if (sink instanceof Source) {
 213                                         currentComponents.add(sink);
 214                                 }
 215                         }
 216                 }
 217                 return components.build();
 218         }
 219
 220         //
 221         // STATIC METHODS
 222         //
 223
 224         /**
 225          * Returns a new pipeline builder.
 226          *
 227          * @param source
 228          *              The source at which to start
 229          * @return A builder for a new pipeline
 230          */
 231         public static Builder builder(Source source) {
 232                 return new Builder(source);
 233         }
 234
 235         /**
 236          * A builder for a {@link Pipeline}.
 237          *
 238          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 239          */
 240         public static class Builder {
 241
 242                 /** The source of the pipeline. */
 243                 private final Source source;
 244
 245                 /** The sinks to which each source streams. */
 246                 private Multimap<Source, Sink> nextSinks = ArrayListMultimap.create();
 247
 248                 /** The last added source. */
 249                 private Source lastSource;
 250
 251                 /**
 252                  * Creates a new builder.
 253                  *
 254                  * @param source
 255                  *              The source that starts the pipeline
 256                  */
 257                 private Builder(Source source) {
 258                         this.source = source;
 259                         lastSource = source;
 260                 }
 261
 262                 /**
 263                  * Adds a {@link Sink} (or {@link Filter} as a recipient for the last added
 264                  * {@link Source}.
 265                  *
 266                  * @param sink
 267                  *              The sink to add
 268                  * @return This builder
 269                  * @throws IllegalStateException
 270                  *              if the last added {@link Sink} was not also a {@link Source}
 271                  */
 272                 public Builder to(Sink sink) {
 273                         Preconditions.checkState(lastSource != null, "last added Sink was not a Source");
 274                         nextSinks.put(lastSource, sink);
 275                         lastSource = (sink instanceof Filter) ? (Source) sink : null;
 276                         return this;
 277                 }
 278
 279                 /**
 280                  * Locates the given source and sets it as the last added node so that the
 281                  * next invocation of {@link #to(Sink)} can “fork” the pipeline.
 282                  *
 283                  * @param source
 284                  *              The source to locate
 285                  * @return This builder
 286                  * @throws IllegalStateException
 287                  *              if the given source was not previously added as a sink
 288                  */
 289                 public Builder find(Source source) {
 290                         Preconditions.checkState(nextSinks.containsValue(source));
 291                         lastSource = source;
 292                         return this;
 293                 }
 294
 295                 /**
 296                  * Builds the pipeline.
 297                  *
 298                  * @return The created pipeline
 299                  */
 300                 public Pipeline build() {
 301                         return new Pipeline(source, ImmutableMultimap.copyOf(nextSinks));
 302                 }
 303
 304         }
 305
 306         /**
 307          * A connection is responsible for streaming audio from one {@link Source} to
 308          * an arbitrary number of {@link Sink}s it is connected to. A connection is
 309          * started by creating a {@link Thread} wrapping it and starting said thread.
 310          *
 311          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 312          */
 313         public class Connection implements Runnable {
 314
 315                 /** The source. */
 316                 private final Source source;
 317
 318                 /** The sinks. */
 319                 private final Collection<Sink> sinks;
 320
 321                 /** Whether the feeder was stopped. */
 322                 private final AtomicBoolean stopped = new AtomicBoolean(false);
 323
 324                 /** The executor service. */
 325                 private final ExecutorService executorService;
 326
 327                 /** The number of copied bytes. */
 328                 private long counter;
 329
 330                 /**
 331                  * Creates a new connection.
 332                  *
 333                  * @param source
 334                  *              The source of the stream
 335                  * @param sinks
 336                  *              The sinks to which to stream
 337                  */
 338                 public Connection(Source source, Collection<Sink> sinks) {
 339                         this.source = source;
 340                         this.sinks = sinks;
 341                         if (sinks.size() == 1) {
 342                                 executorService = MoreExecutors.sameThreadExecutor();
 343                         } else {
 344                                 executorService = Executors.newCachedThreadPool();
 345                         }
 346                 }
 347
 348                 //
 349                 // ACCESSORS
 350                 //
 351
 352                 /**
 353                  * Returns the number of bytes that this connection has received from its
 354                  * source during its lifetime.
 355                  *
 356                  * @return The number of processed input bytes
 357                  */
 358                 public long counter() {
 359                         return counter;
 360                 }
 361
 362                 //
 363                 // ACTIONS
 364                 //
 365
 366                 /** Stops this connection. */
 367                 public void stop() {
 368                         stopped.set(true);
 369                 }
 370
 371                 //
 372                 // RUNNABLE METHODS
 373                 //
 374
 375                 @Override
 376                 public void run() {
 377                         Metadata firstMetadata = null;
 378                         while (!stopped.get()) {
 379                                 try {
 380                                         final byte[] buffer;
 381                                         try {
 382                                                 logger.finest(String.format("Getting %d bytes from %s...", 4096, source));
 383                                                 buffer = source.get(4096);
 384                                                 logger.finest(String.format("Got %d bytes from %s.", buffer.length, source));
 385                                         } catch (IOException ioe1) {
 386                                                 throw new IOException(String.format("I/O error while reading from %s.", source), ioe1);
 387                                         }
 388                                         List<Future<Void>> futures = executorService.invokeAll(FluentIterable.from(sinks).transform(new Function<Sink, Callable<Void>>() {
 389
 390                                                 @Override
 391                                                 public Callable<Void> apply(final Sink sink) {
 392                                                         return new Callable<Void>() {
 393
 394                                                                 @Override
 395                                                                 public Void call() throws Exception {
 396                                                                         try {
 397                                                                                 logger.finest(String.format("Sending %d bytes to %s.", buffer.length, sink));
 398                                                                                 sink.process(buffer);
 399                                                                                 logger.finest(String.format("Sent %d bytes to %s.", buffer.length, sink));
 400                                                                         } catch (IOException ioe1) {
 401                                                                                 throw new IOException(String.format("I/O error while writing to %s", sink), ioe1);
 402                                                                         }
 403                                                                         return null;
 404                                                                 }
 405                                                         };
 406                                                 }
 407                                         }).toList());
 408                                         /* check all threads for exceptions. */
 409                                         for (Future<Void> future : futures) {
 410                                                 future.get();
 411                                         }
 412                                         counter += buffer.length;
 413                                 } catch (IOException e) {
 414                                         /* TODO */
 415                                         e.printStackTrace();
 416                                         break;
 417                                 } catch (InterruptedException e) {
 418                                         /* TODO */
 419                                         e.printStackTrace();
 420                                         break;
 421                                 } catch (ExecutionException e) {
 422                                         /* TODO */
 423                                         e.printStackTrace();
 424                                         break;
 425                                 }
 426                         }
 427                 }
 428
 429         }
 430
 431         /**
 432          * Container for input and output counters.
 433          *
 434          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 435          */
 436         public static class TrafficCounter {
 437
 438                 /** The number of input bytes. */
 439                 private final long input;
 440
 441                 /** The number of output bytes. */
 442                 private final long output;
 443
 444                 /**
 445                  * Creates a new traffic counter.
 446                  *
 447                  * @param input
 448                  *              The number of input bytes (may be {@code -1} to signify non-available
 449                  *              input)
 450                  * @param output
 451                  *              The number of output bytes (may be {@code -1} to signify non-available
 452                  *              output)
 453                  */
 454                 public TrafficCounter(long input, long output) {
 455                         this.input = input;
 456                         this.output = output;
 457                 }
 458
 459                 //
 460                 // ACCESSORS
 461                 //
 462
 463                 /**
 464                  * Returns the number of input bytes.
 465                  *
 466                  * @return The number of input bytes, or {@link Optional#absent()} if the
 467                  *         component can not receive input
 468                  */
 469                 public Optional<Long> input() {
 470                         return (input == -1) ? Optional.<Long>absent() : Optional.of(input);
 471                 }
 472
 473                 /**
 474                  * Returns the number of output bytes.
 475                  *
 476                  * @return The number of output bytes, or {@link Optional#absent()} if the
 477                  *         component can not send output
 478                  */
 479                 public Optional<Long> output() {
 480                         return (output == -1) ? Optional.<Long>absent() : Optional.of(output);
 481                 }
 482
 483         }
 484
 485 }