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.name()));
  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 as %s.", component.name(), controlledComponent.name(), metadata));
  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                                 logger.info(String.format("Opening %s with %s...", sink.name(), source.metadata()));
 163                                 sink.open(source.metadata());
 164                                 if (sink instanceof Filter) {
 165                                         sources.add((Source) sink);
 166                                 }
 167                         }
 168                 }
 169                 for (Connection connection : connections) {
 170                         logger.info(String.format("Starting Connection from %s to %s.", connection.source.name(), FluentIterable.from(connection.sinks).transform(new Function<Sink, String>() {
 171                                 @Override
 172                                 public String apply(Sink sink) {
 173                                         return sink.name();
 174                                 }
 175                         })));
 176                         new Thread(connection).start();
 177                 }
 178         }
 179
 180         public void stop() {
 181                 if (!connections.isEmpty()) {
 182                         /* pipeline is not running. */
 183                         return;
 184                 }
 185                 for (Connection connection : connections) {
 186                         connection.stop();
 187                 }
 188         }
 189
 190         //
 191         // ITERABLE METHODS
 192         //
 193
 194         @Override
 195         public Iterator<ControlledComponent> iterator() {
 196                 return components().iterator();
 197         }
 198
 199         //
 200         // PRIVATE METHODS
 201         //
 202
 203         /**
 204          * Returns all components of this pipeline, listed breadth-first, starting with
 205          * the source.
 206          *
 207          * @return All components of this pipeline
 208          */
 209         public List<ControlledComponent> components() {
 210                 ImmutableList.Builder<ControlledComponent> components = ImmutableList.builder();
 211                 List<ControlledComponent> currentComponents = Lists.newArrayList();
 212                 components.add(source);
 213                 currentComponents.add(source);
 214                 while (!currentComponents.isEmpty()) {
 215                         Collection<Sink> sinks = this.sinks((Source) currentComponents.remove(0));
 216                         for (Sink sink : sinks) {
 217                                 components.add(sink);
 218                                 if (sink instanceof Source) {
 219                                         currentComponents.add(sink);
 220                                 }
 221                         }
 222                 }
 223                 return components.build();
 224         }
 225
 226         //
 227         // STATIC METHODS
 228         //
 229
 230         /**
 231          * Returns a new pipeline builder.
 232          *
 233          * @param source
 234          *              The source at which to start
 235          * @return A builder for a new pipeline
 236          */
 237         public static Builder builder(Source source) {
 238                 return new Builder(source);
 239         }
 240
 241         /**
 242          * A builder for a {@link Pipeline}.
 243          *
 244          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 245          */
 246         public static class Builder {
 247
 248                 /** The source of the pipeline. */
 249                 private final Source source;
 250
 251                 /** The sinks to which each source streams. */
 252                 private Multimap<Source, Sink> nextSinks = ArrayListMultimap.create();
 253
 254                 /** The last added source. */
 255                 private Source lastSource;
 256
 257                 /**
 258                  * Creates a new builder.
 259                  *
 260                  * @param source
 261                  *              The source that starts the pipeline
 262                  */
 263                 private Builder(Source source) {
 264                         this.source = source;
 265                         lastSource = source;
 266                 }
 267
 268                 /**
 269                  * Adds a {@link Sink} (or {@link Filter} as a recipient for the last added
 270                  * {@link Source}.
 271                  *
 272                  * @param sink
 273                  *              The sink to add
 274                  * @return This builder
 275                  * @throws IllegalStateException
 276                  *              if the last added {@link Sink} was not also a {@link Source}
 277                  */
 278                 public Builder to(Sink sink) {
 279                         Preconditions.checkState(lastSource != null, "last added Sink was not a Source");
 280                         nextSinks.put(lastSource, sink);
 281                         lastSource = (sink instanceof Filter) ? (Source) sink : null;
 282                         return this;
 283                 }
 284
 285                 /**
 286                  * Locates the given source and sets it as the last added node so that the
 287                  * next invocation of {@link #to(Sink)} can “fork” the pipeline.
 288                  *
 289                  * @param source
 290                  *              The source to locate
 291                  * @return This builder
 292                  * @throws IllegalStateException
 293                  *              if the given source was not previously added as a sink
 294                  */
 295                 public Builder find(Source source) {
 296                         Preconditions.checkState(nextSinks.containsValue(source));
 297                         lastSource = source;
 298                         return this;
 299                 }
 300
 301                 /**
 302                  * Builds the pipeline.
 303                  *
 304                  * @return The created pipeline
 305                  */
 306                 public Pipeline build() {
 307                         return new Pipeline(source, ImmutableMultimap.copyOf(nextSinks));
 308                 }
 309
 310         }
 311
 312         /**
 313          * A connection is responsible for streaming audio from one {@link Source} to
 314          * an arbitrary number of {@link Sink}s it is connected to. A connection is
 315          * started by creating a {@link Thread} wrapping it and starting said thread.
 316          *
 317          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 318          */
 319         public class Connection implements Runnable {
 320
 321                 /** The source. */
 322                 private final Source source;
 323
 324                 /** The sinks. */
 325                 private final Collection<Sink> sinks;
 326
 327                 /** Whether the feeder was stopped. */
 328                 private final AtomicBoolean stopped = new AtomicBoolean(false);
 329
 330                 /** The executor service. */
 331                 private final ExecutorService executorService;
 332
 333                 /** The number of copied bytes. */
 334                 private long counter;
 335
 336                 /**
 337                  * Creates a new connection.
 338                  *
 339                  * @param source
 340                  *              The source of the stream
 341                  * @param sinks
 342                  *              The sinks to which to stream
 343                  */
 344                 public Connection(Source source, Collection<Sink> sinks) {
 345                         this.source = source;
 346                         this.sinks = sinks;
 347                         if (sinks.size() == 1) {
 348                                 executorService = MoreExecutors.sameThreadExecutor();
 349                         } else {
 350                                 executorService = Executors.newCachedThreadPool();
 351                         }
 352                 }
 353
 354                 //
 355                 // ACCESSORS
 356                 //
 357
 358                 /**
 359                  * Returns the number of bytes that this connection has received from its
 360                  * source during its lifetime.
 361                  *
 362                  * @return The number of processed input bytes
 363                  */
 364                 public long counter() {
 365                         return counter;
 366                 }
 367
 368                 //
 369                 // ACTIONS
 370                 //
 371
 372                 /** Stops this connection. */
 373                 public void stop() {
 374                         stopped.set(true);
 375                 }
 376
 377                 //
 378                 // RUNNABLE METHODS
 379                 //
 380
 381                 @Override
 382                 public void run() {
 383                         Metadata firstMetadata = null;
 384                         while (!stopped.get()) {
 385                                 try {
 386                                         final byte[] buffer;
 387                                         try {
 388                                                 logger.finest(String.format("Getting %d bytes from %s...", 4096, source));
 389                                                 buffer = source.get(4096);
 390                                                 logger.finest(String.format("Got %d bytes from %s.", buffer.length, source));
 391                                         } catch (IOException ioe1) {
 392                                                 throw new IOException(String.format("I/O error while reading from %s.", source), ioe1);
 393                                         }
 394                                         List<Future<Void>> futures = executorService.invokeAll(FluentIterable.from(sinks).transform(new Function<Sink, Callable<Void>>() {
 395
 396                                                 @Override
 397                                                 public Callable<Void> apply(final Sink sink) {
 398                                                         return new Callable<Void>() {
 399
 400                                                                 @Override
 401                                                                 public Void call() throws Exception {
 402                                                                         try {
 403                                                                                 logger.finest(String.format("Sending %d bytes to %s.", buffer.length, sink));
 404                                                                                 sink.process(buffer);
 405                                                                                 logger.finest(String.format("Sent %d bytes to %s.", buffer.length, sink));
 406                                                                         } catch (IOException ioe1) {
 407                                                                                 throw new IOException(String.format("I/O error while writing to %s", sink), ioe1);
 408                                                                         }
 409                                                                         return null;
 410                                                                 }
 411                                                         };
 412                                                 }
 413                                         }).toList());
 414                                         /* check all threads for exceptions. */
 415                                         for (Future<Void> future : futures) {
 416                                                 future.get();
 417                                         }
 418                                         counter += buffer.length;
 419                                 } catch (IOException e) {
 420                                         /* TODO */
 421                                         e.printStackTrace();
 422                                         break;
 423                                 } catch (InterruptedException e) {
 424                                         /* TODO */
 425                                         e.printStackTrace();
 426                                         break;
 427                                 } catch (ExecutionException e) {
 428                                         /* TODO */
 429                                         e.printStackTrace();
 430                                         break;
 431                                 }
 432                         }
 433                 }
 434
 435         }
 436
 437         /**
 438          * Container for input and output counters.
 439          *
 440          * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
 441          */
 442         public static class TrafficCounter {
 443
 444                 /** The number of input bytes. */
 445                 private final long input;
 446
 447                 /** The number of output bytes. */
 448                 private final long output;
 449
 450                 /**
 451                  * Creates a new traffic counter.
 452                  *
 453                  * @param input
 454                  *              The number of input bytes (may be {@code -1} to signify non-available
 455                  *              input)
 456                  * @param output
 457                  *              The number of output bytes (may be {@code -1} to signify non-available
 458                  *              output)
 459                  */
 460                 public TrafficCounter(long input, long output) {
 461                         this.input = input;
 462                         this.output = output;
 463                 }
 464
 465                 //
 466                 // ACCESSORS
 467                 //
 468
 469                 /**
 470                  * Returns the number of input bytes.
 471                  *
 472                  * @return The number of input bytes, or {@link Optional#absent()} if the
 473                  *         component can not receive input
 474                  */
 475                 public Optional<Long> input() {
 476                         return (input == -1) ? Optional.<Long>absent() : Optional.of(input);
 477                 }
 478
 479                 /**
 480                  * Returns the number of output bytes.
 481                  *
 482                  * @return The number of output bytes, or {@link Optional#absent()} if the
 483                  *         component can not send output
 484                  */
 485                 public Optional<Long> output() {
 486                         return (output == -1) ? Optional.<Long>absent() : Optional.of(output);
 487                 }
 488
 489         }
 490
 491 }