Motown in-proc streaming framework used to implement media services.

services/media/framework/engine.h

+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef SERVICES_MEDIA_FRAMEWORK_ENGINE_H_
+#define SERVICES_MEDIA_FRAMEWORK_ENGINE_H_
+
+#include <deque>
+#include <list>
+
+#include "services/media/framework/stages/active_sink_stage.h"
+#include "services/media/framework/stages/active_source_stage.h"
+#include "services/media/framework/stages/distributor_stage.h"
+#include "services/media/framework/stages/lpcm_transform_stage.h"
+#include "services/media/framework/stages/packet_transform_stage.h"
+#include "services/media/framework/stages/stage.h"
+
+namespace mojo {
+namespace media {
+
+// Host for a source, sink or transform.
+class Engine {
+ public:
+  class Input;
+  class Output;
+
+  // Opaque Stage pointer used for graph building.
+  class Part {
+   public:
+    Part() : stage_(nullptr) {}
+
+    uint32_t input_count();
+    Input input(uint32_t index);

[johngro, 2016/01/20 00:25:32]

re: the accessors here which are making copies of Inputs, Outputs and Parts

Is there an advantage to doing this?  Why not just return a const& and let the
caller make a copy using the default copy constructor if they need a mutable
copy?

[dalesat, 2016/01/25 17:47:29]

The intent is that Part, Input and Output are value types comparable to
iterators. Part is pointer-sized and there's no benefit to passing it by
reference. Input and Output are pointer + 32 bits, so your suggestion has some
benefit for those. Regardless, I don't want to introduce any concerns about
lifetime.

[dalesat, 2016/01/25 23:29:37]

Also, Input and Output are synthesized by these methods (from a stage pointer
and an input/output index). There's nothing to reference if we wanted to return
a reference.

[johngro, 2016/01/27 22:35:22]

Acknowledged.

+    Input input();
+    uint32_t output_count();
+    Output output(uint32_t index);
+    Output output();
+    Part upstream_part(uint32_t index);
+    Part upstream_part();
+    Part downstream_part(uint32_t index);
+    Part downstream_part();
+
+   private:
+    Part(Stage* stage) : stage_(stage) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Stage* stage_;
+
+    friend Engine;
+    friend Input;
+    friend Output;
+  };
+
+  // Opaque StageInput pointer used for graph building.
+  class Input {
+   public:
+    Input() : stage_(nullptr), index_(0) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Part part() { return Part(stage_); }
+
+    bool connected() {
+      DCHECK(stage_);
+      return stage_input().upstream_stage() != nullptr;
+    }
+
+    Part upstream_part() {
+      DCHECK(connected());
+      return Part(stage_input().upstream_stage());
+    }
+
+   private:
+    Input(Stage* stage, uint32_t index) :
+        stage_(stage), index_(index) {
+      DCHECK(stage_);
+      DCHECK(index_ < stage_->input_count());
+    }
+
+    StageInput& stage_input() {
+      DCHECK(stage_);
+      return stage_->input(index_);
+    }
+
+    Stage* stage_;
+    uint32_t index_;
+
+    friend Engine;
+    friend Part;
+    friend Output;
+  };
+
+  // Opaque StageOutput pointer used for graph building.
+  class Output {
+   public:
+    Output() : stage_(nullptr), index_(0) {}
+
+    explicit operator bool() const { return stage_ != nullptr; }
+
+    Part part() { return Part(stage_); }
+
+    bool connected() {
+      DCHECK(stage_);
+      return stage_output().downstream_stage() != nullptr;
+    }
+
+    Part downstream_part() {
+      DCHECK(connected());
+      return Part(stage_output().downstream_stage());
+    }
+
+   private:
+    Output(Stage* stage, uint32_t index) :
+        stage_(stage), index_(index) {
+      DCHECK(stage_);
+      DCHECK(index_ < stage_->output_count());
+    }
+
+    StageOutput& stage_output() {
+      DCHECK(stage_);
+      return stage_->output(index_);
+    }
+
+    Stage* stage_;
+    uint32_t index_;
+
+    friend Engine;
+    friend Part;
+    friend Input;
+  };
+
+  Engine();
+
+  ~Engine();
+
+  // Adds a part to the engine.
+  template<typename T, typename TBase>
+  Part Add(SharedPtr<T, TBase> t) {
+    DCHECK(t);
+    return Add(CreateStage(std::shared_ptr<TBase>(t)));
+  }
+
+  // Removes a part from the engine after disconnecting it from other parts.
+  void Remove(Part part);
+
+  // Connects an output connector to an input connector. Returns the dowstream
+  // part.
+  Part Connect(
+      Output output,
+      Input input);
+
+  // Connects a part with exactly one output to a part with exactly one input.
+  // Returns the downstream part.
+  Part Connect(Part upstream_part, Part downstream_part);
+
+  // Connects an output connector to a part that has exactly one input. Returns
+  // the downstream part.
+  Part Connect(
+      Output output,
+      Part downstream_part);
+
+  // Connects a part with exactly one output to an input connector. Returns the
+  // downstream part.
+  Part Connect(
+      Part upstream_part,
+      Input input);
+
+  // Disconnects an output connector and the input connector to which it's
+  // connected.
+  void Disconnect(Output output);
+
+  // Disconnects an input connector and the output connector to which it's
+  // connected.
+  void Disconnect(Input input);
+
+  // Disconnects and removes part and everything connected to it.
+  void RemoveAll(Part part);
+
+  // Disconnects and removes everything connected to output.
+  void RemoveAll(Output output);
+
+  // Disconnects and removes everything connected to input.
+  void RemoveAll(Input input);
+
+  // Adds all the parts in t (which must all have one input and one output) and
+  // connects them in sequence to the output connector. Returns the output
+  // connector of the last part or the output parameter if t is empty.
+  template<typename T>
+  Output AddAndConnectAll(
+      Output output,
+      const T& t) {
+    for (auto& element : t) {
+      Part part = Add(CreateStage(element));
+      Connect(output, part.input());
+      output = part.output();
+    }
+    return output;
+  }
+
+  // Prepares the engine.
+  void Prepare();
+
+  // Removes all parts from the engine.
+  void reset();

[johngro, 2016/01/20 00:25:32]

Reset

[dalesat, 2016/01/25 23:29:37]

Done.

+
+  // Pushes the stage to the supply backlog if it isn't already there.
+  void PushToSupplyBacklog(Stage* stage);
+
+  // Pushes the stage to the demand backlog if it isn't already there.
+  void PushToDemandBacklog(Stage* stage);
+
+ private:
+  // Adds a stage to the engine.
+  Part Add(Stage* stage);
+
+  // Creates a stage from a source, sink or transform. A specialization of this
+  // template is defined for each type of source, sink or transform that can be
+  // added to the engine.
+  template<typename T>
+  static Stage* CreateStage(std::shared_ptr<T> t);
+
+  // CreateStage template specialization for MultiStreamPacketSource.
+  static Stage* CreateStage(MultiStreamPacketSourcePtr source);
+
+  // CreateStage template specialization for PacketTransform.
+  static Stage* CreateStage(PacketTransformPtr transform);
+
+  // CreateStage template specialization for ActiveSource.
+  static Stage* CreateStage(ActiveSourcePtr source);
+
+  // CreateStage template specialization for ActiveSink.
+  static Stage* CreateStage(ActiveSinkPtr sink);
+
+  // CreateStage template specialization for LpcmTransform.
+  static Stage* CreateStage(LpcmTransformPtr transform);
+
+  // Prepares a stage.
+  void Prepare(Stage* stage);
+
+  // Processes the entire backlog.
+  void Update();
+
+  // Performs processing for a single stage, updating the backlog accordingly.
+  void Update(Stage *stage);
+
+  // Pops a stage from the supply backlog and returns it or returns nullptr if
+  // the supply backlog is empty.
+  Stage* PopFromSupplyBacklog();
+
+  // Pops a stage from the demand backlog and returns it or returns nullptr if
+  // the demand backlog is empty.
+  Stage* PopFromDemandBacklog();
+
+  std::list<std::unique_ptr<Stage>> stages_;
+  std::list<Stage*> sources_;
+  std::list<Stage*> sinks_;
+  std::deque<Stage*> supply_backlog_;
+  std::deque<Stage*> demand_backlog_;
+  Stage::UpdateCallback update_function_;
+  bool packets_produced_;
+};
+
+}  // namespace media
+}  // namespace mojo
+
+#endif  // SERVICES_MEDIA_FRAMEWORK_ENGINE_ENGINE_H_