Project import generated by Copybara.

GitOrigin-RevId: ff83882955f1a1e2a043ff4e71278be9d7217bbe

mediapipe/framework/BUILD

@@ -15,6 +15,7 @@
 #
 
 load("//mediapipe/framework/port:build_config.bzl", "mediapipe_proto_library")
+load("@bazel_skylib//:bzl_library.bzl", "bzl_library")
 
 licenses(["notice"])
 
@@ -27,10 +28,28 @@ package_group(
     ],
 )
 
-exports_files([
-    "transitive_protos.bzl",
-    "encode_binary_proto.bzl",
-])
+bzl_library(
+    name = "transitive_protos_bzl",
+    srcs = [
+        "transitive_protos.bzl",
+    ],
+    visibility = ["//mediapipe/framework:__subpackages__"],
+)
+
+bzl_library(
+    name = "encode_binary_proto_bzl",
+    srcs = [
+        "encode_binary_proto.bzl",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+alias(
+    name = "encode_binary_proto",
+    actual = ":encode_binary_proto_bzl",
+    deprecation = "Use encode_binary_proto_bzl",
+    visibility = ["//visibility:public"],
+)
 
 mediapipe_proto_library(
     name = "calculator_proto",

mediapipe/framework/api2/BUILD

@@ -1,15 +1,8 @@
 package(
-    default_visibility = [":preview_users"],
+    default_visibility = ["//visibility:public"],
     features = ["-use_header_modules"],
 )
 
-package_group(
-    name = "preview_users",
-    packages = [
-        "//mediapipe/...",
-    ],
-)
-
 licenses(["notice"])
 
 cc_library(

mediapipe/framework/calculator.proto

@@ -422,6 +422,9 @@ message CalculatorGraphConfig {
   // the graph config.
   string type = 20;
 
-  // Can be used for annotating a graph.
+  // The types and default values for graph options, in proto2 syntax.
   MediaPipeOptions options = 1001;
+
+  // The types and default values for graph options, in proto3 syntax.
+  repeated google.protobuf.Any graph_options = 1002;
 }

mediapipe/framework/calculator_graph.cc

@@ -411,7 +411,8 @@ absl::Status CalculatorGraph::Initialize(
 
 absl::Status CalculatorGraph::ObserveOutputStream(
     const std::string& stream_name,
-    std::function<absl::Status(const Packet&)> packet_callback) {
+    std::function<absl::Status(const Packet&)> packet_callback,
+    bool observe_timestamp_bounds) {
   RET_CHECK(initialized_).SetNoLogging()
       << "CalculatorGraph is not initialized.";
   // TODO Allow output observers to be attached by graph level
@@ -425,7 +426,7 @@ absl::Status CalculatorGraph::ObserveOutputStream(
   auto observer = absl::make_unique<internal::OutputStreamObserver>();
   MP_RETURN_IF_ERROR(observer->Initialize(
       stream_name, &any_packet_type_, std::move(packet_callback),
-      &output_stream_managers_[output_stream_index]));
+      &output_stream_managers_[output_stream_index], observe_timestamp_bounds));
   graph_output_streams_.push_back(std::move(observer));
   return absl::OkStatus();
 }

mediapipe/framework/calculator_graph.h

@@ -157,7 +157,8 @@ class CalculatorGraph {
   // TODO: Rename to AddOutputStreamCallback.
   absl::Status ObserveOutputStream(
       const std::string& stream_name,
-      std::function<absl::Status(const Packet&)> packet_callback);
+      std::function<absl::Status(const Packet&)> packet_callback,
+      bool observe_timestamp_bounds = false);
 
   // Adds an OutputStreamPoller for a stream. This provides a synchronous,
   // polling API for accessing a stream's output. Should only be called before

mediapipe/framework/calculator_graph_bounds_test.cc

@@ -1518,5 +1518,72 @@ TEST(CalculatorGraphBoundsTest, OffsetAndBound) {
   MP_ASSERT_OK(graph.WaitUntilDone());
 }
 
+// A Calculator that sends empty output stream packets.
+class EmptyPacketCalculator : public CalculatorBase {
+ public:
+  static absl::Status GetContract(CalculatorContract* cc) {
+    cc->Inputs().Index(0).Set<int>();
+    cc->Outputs().Index(0).SetSameAs(&cc->Inputs().Index(0));
+    return absl::OkStatus();
+  }
+  absl::Status Open(CalculatorContext* cc) final { return absl::OkStatus(); }
+  absl::Status Process(CalculatorContext* cc) final {
+    if (cc->InputTimestamp().Value() % 2 == 0) {
+      cc->Outputs().Index(0).AddPacket(Packet().At(cc->InputTimestamp()));
+    }
+    return absl::OkStatus();
+  }
+};
+REGISTER_CALCULATOR(EmptyPacketCalculator);
+
+// This test shows that an output timestamp bound can be specified by outputing
+// an empty packet with a settled timestamp.
+TEST(CalculatorGraphBoundsTest, EmptyPacketOutput) {
+  // OffsetAndBoundCalculator runs on parallel threads and sends ts
+  // occasionally.
+  std::string config_str = R"(
+            input_stream: "input_0"
+            node {
+              calculator: "EmptyPacketCalculator"
+              input_stream: "input_0"
+              output_stream: "empty_0"
+            }
+            node {
+              calculator: "ProcessBoundToPacketCalculator"
+              input_stream: "empty_0"
+              output_stream: "output_0"
+            }
+          )";
+  CalculatorGraphConfig config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(config_str);
+  CalculatorGraph graph;
+  std::vector<Packet> output_0_packets;
+  MP_ASSERT_OK(graph.Initialize(config));
+  MP_ASSERT_OK(graph.ObserveOutputStream("output_0", [&](const Packet& p) {
+    output_0_packets.push_back(p);
+    return absl::OkStatus();
+  }));
+  MP_ASSERT_OK(graph.StartRun({}));
+  MP_ASSERT_OK(graph.WaitUntilIdle());
+
+  // Send in packets.
+  for (int i = 0; i < 9; ++i) {
+    const int ts = 10 + i * 10;
+    Packet p = MakePacket<int>(i).At(Timestamp(ts));
+    MP_ASSERT_OK(graph.AddPacketToInputStream("input_0", p));
+    MP_ASSERT_OK(graph.WaitUntilIdle());
+  }
+
+  // 9 empty packets are converted to bounds and then to packets.
+  EXPECT_EQ(output_0_packets.size(), 9);
+  for (int i = 0; i < 9; ++i) {
+    EXPECT_EQ(output_0_packets[i].Timestamp(), Timestamp(10 + i * 10));
+  }
+
+  // Shutdown the graph.
+  MP_ASSERT_OK(graph.CloseAllPacketSources());
+  MP_ASSERT_OK(graph.WaitUntilDone());
+}
+
 }  // namespace
 }  // namespace mediapipe

mediapipe/framework/deps/BUILD

@@ -16,11 +16,20 @@
 #   The dependencies of mediapipe.
 
 load("//mediapipe/framework/port:build_config.bzl", "mediapipe_cc_proto_library")
+load("@bazel_skylib//:bzl_library.bzl", "bzl_library")
 
 licenses(["notice"])
 
 package(default_visibility = ["//visibility:private"])
 
+bzl_library(
+    name = "expand_template_bzl",
+    srcs = [
+        "expand_template.bzl",
+    ],
+    visibility = ["//mediapipe/framework:__subpackages__"],
+)
+
 proto_library(
     name = "proto_descriptor_proto",
     srcs = ["proto_descriptor.proto"],

mediapipe/framework/formats/BUILD

@@ -295,6 +295,7 @@ cc_library(
         "//mediapipe/framework/formats:image_format_cc_proto",
         "@com_google_absl//absl/synchronization",
         "//mediapipe/framework:port",
+        "//mediapipe/framework:type_map",
         "//mediapipe/framework/port:logging",
     ] + select({
         "//conditions:default": [

mediapipe/framework/formats/image.cc

@@ -14,6 +14,8 @@
 
 #include "mediapipe/framework/formats/image.h"
 
+#include "mediapipe/framework/type_map.h"
+
 namespace mediapipe {
 
 // TODO Refactor common code from GpuBufferToImageFrameCalculator
@@ -67,8 +69,7 @@ bool Image::ConvertToGpu() const {
 #else
   if (use_gpu_) return true;  // Already on GPU.
 #if MEDIAPIPE_GPU_BUFFER_USE_CV_PIXEL_BUFFER
-  auto packet = MakePacket<ImageFrame>(std::move(*image_frame_));
-  image_frame_ = nullptr;
+  auto packet = PointToForeign<ImageFrame>(image_frame_.get());
   CFHolder<CVPixelBufferRef> buffer;
   auto status = CreateCVPixelBufferForImageFramePacket(packet, true, &buffer);
   CHECK_OK(status);
@@ -94,4 +95,7 @@ bool Image::ConvertToGpu() const {
 #endif  // MEDIAPIPE_DISABLE_GPU
 }
 
+MEDIAPIPE_REGISTER_TYPE(mediapipe::Image, "::mediapipe::Image", nullptr,
+                        nullptr);
+
 }  // namespace mediapipe

mediapipe/framework/formats/image.h

@@ -72,8 +72,8 @@ class Image {
 
   // Creates an Image representing the same image content as the ImageFrame
   // the input shared pointer points to, and retaining shared ownership.
-  explicit Image(ImageFrameSharedPtr frame_buffer)
-      : image_frame_(std::move(frame_buffer)) {
+  explicit Image(ImageFrameSharedPtr image_frame)
+      : image_frame_(std::move(image_frame)) {
     use_gpu_ = false;
     pixel_mutex_ = std::make_shared<absl::Mutex>();
   }

mediapipe/framework/formats/tensor.cc

@@ -30,6 +30,9 @@
 
 namespace mediapipe {
 
+// Zero and negative values are not checked here.
+bool IsPowerOfTwo(int v) { return (v & (v - 1)) == 0; }
+
 int BhwcBatchFromShape(const Tensor::Shape& shape) {
   LOG_IF(FATAL, shape.dims.empty())
       << "Tensor::Shape must be non-empty to retrieve a named dimension";
@@ -237,6 +240,12 @@ void Tensor::AllocateOpenGlTexture2d() const {
       glTexStorage2D(GL_TEXTURE_2D, 1, GL_RGBA32F, texture_width_,
                      texture_height_);
     } else {
+      // GLES2.0 supports only clamp addressing mode for NPOT textures.
+      // If any of dimensions is NPOT then both addressing modes are clamp.
+      if (!IsPowerOfTwo(texture_width_) || !IsPowerOfTwo(texture_height_)) {
+        glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_WRAP_S, GL_CLAMP_TO_EDGE);
+        glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_WRAP_T, GL_CLAMP_TO_EDGE);
+      }
       // We assume all contexts will have the same extensions, so we only check
       // once for OES_texture_float extension, to save time.
       static bool has_oes_extension =

mediapipe/framework/graph_output_stream.cc

@@ -14,13 +14,16 @@
 
 #include "mediapipe/framework/graph_output_stream.h"
 
+#include "absl/synchronization/mutex.h"
+#include "mediapipe/framework/port/status.h"
+
 namespace mediapipe {
 
 namespace internal {
 
 absl::Status GraphOutputStream::Initialize(
     const std::string& stream_name, const PacketType* packet_type,
-    OutputStreamManager* output_stream_manager) {
+    OutputStreamManager* output_stream_manager, bool observe_timestamp_bounds) {
   RET_CHECK(output_stream_manager);
 
   // Initializes input_stream_handler_ with one input stream as the observer.
@@ -31,6 +34,7 @@ absl::Status GraphOutputStream::Initialize(
   input_stream_handler_ = absl::make_unique<GraphOutputStreamHandler>(
       tag_map, /*cc_manager=*/nullptr, MediaPipeOptions(),
       /*calculator_run_in_parallel=*/false);
+  input_stream_handler_->SetProcessTimestampBounds(observe_timestamp_bounds);
   const CollectionItemId& id = tag_map->BeginId();
   input_stream_ = absl::make_unique<InputStreamManager>();
   MP_RETURN_IF_ERROR(
@@ -52,20 +56,58 @@ void GraphOutputStream::PrepareForRun(
 absl::Status OutputStreamObserver::Initialize(
     const std::string& stream_name, const PacketType* packet_type,
     std::function<absl::Status(const Packet&)> packet_callback,
-    OutputStreamManager* output_stream_manager) {
+    OutputStreamManager* output_stream_manager, bool observe_timestamp_bounds) {
   RET_CHECK(output_stream_manager);
 
   packet_callback_ = std::move(packet_callback);
+  observe_timestamp_bounds_ = observe_timestamp_bounds;
   return GraphOutputStream::Initialize(stream_name, packet_type,
-                                       output_stream_manager);
+                                       output_stream_manager,
+                                       observe_timestamp_bounds);
 }
 
 absl::Status OutputStreamObserver::Notify() {
+  // Lets one thread perform packets notification as much as possible.
+  // Other threads should quit if a thread is already performing notification.
+  {
+    absl::MutexLock l(&mutex_);
+
+    if (notifying_ == false) {
+      notifying_ = true;
+    } else {
+      return absl::OkStatus();
+    }
+  }
   while (true) {
     bool empty;
     Timestamp min_timestamp = input_stream_->MinTimestampOrBound(&empty);
     if (empty) {
-      break;
+      // Emits an empty packet at timestamp_bound.PreviousAllowedInStream().
+      if (observe_timestamp_bounds_ && min_timestamp < Timestamp::Done()) {
+        Timestamp settled = (min_timestamp == Timestamp::PostStream()
+                                 ? Timestamp::PostStream()
+                                 : min_timestamp.PreviousAllowedInStream());
+        if (last_processed_ts_ < settled) {
+          MP_RETURN_IF_ERROR(packet_callback_(Packet().At(settled)));
+          last_processed_ts_ = settled;
+        }
+      }
+      // Last check to make sure that the min timestamp or bound doesn't change.
+      // If so, flips notifying_ to false to allow any other threads to perform
+      // notification when new packets/timestamp bounds arrive. Otherwise, in
+      // case of the min timestamp or bound getting updated, jumps to the
+      // beginning of the notification loop for a new iteration.
+      {
+        absl::MutexLock l(&mutex_);
+        Timestamp new_min_timestamp =
+            input_stream_->MinTimestampOrBound(&empty);
+        if (new_min_timestamp == min_timestamp) {
+          notifying_ = false;
+          break;
+        } else {
+          continue;
+        }
+      }
     }
     int num_packets_dropped = 0;
     bool stream_is_done = false;
@@ -75,6 +117,7 @@ absl::Status OutputStreamObserver::Notify() {
         << absl::Substitute("Dropped $0 packet(s) on input stream \"$1\".",
                             num_packets_dropped, input_stream_->Name());
     MP_RETURN_IF_ERROR(packet_callback_(packet));
+    last_processed_ts_ = min_timestamp;
   }
   return absl::OkStatus();
 }

mediapipe/framework/graph_output_stream.h

@@ -52,7 +52,8 @@ class GraphOutputStream {
   // is not transferred to the graph output stream object.
   absl::Status Initialize(const std::string& stream_name,
                           const PacketType* packet_type,
-                          OutputStreamManager* output_stream_manager);
+                          OutputStreamManager* output_stream_manager,
+                          bool observe_timestamp_bounds = false);
 
   // Installs callbacks into its GraphOutputStreamHandler.
   virtual void PrepareForRun(std::function<void()> notification_callback,
@@ -99,6 +100,10 @@ class GraphOutputStream {
     }
   };
 
+  bool observe_timestamp_bounds_;
+  absl::Mutex mutex_;
+  bool notifying_ ABSL_GUARDED_BY(mutex_) = false;
+  Timestamp last_processed_ts_ = Timestamp::Unstarted();
   std::unique_ptr<InputStreamHandler> input_stream_handler_;
   std::unique_ptr<InputStreamManager> input_stream_;
 };
@@ -112,7 +117,8 @@ class OutputStreamObserver : public GraphOutputStream {
   absl::Status Initialize(
       const std::string& stream_name, const PacketType* packet_type,
       std::function<absl::Status(const Packet&)> packet_callback,
-      OutputStreamManager* output_stream_manager);
+      OutputStreamManager* output_stream_manager,
+      bool observe_timestamp_bounds = false);
 
   // Notifies the observer of new packets emitted by the observed
   // output stream.
@@ -128,6 +134,7 @@ class OutputStreamObserver : public GraphOutputStream {
 
 // OutputStreamPollerImpl that returns packets to the caller via
 // Next()/NextBatch().
+// TODO: Support observe_timestamp_bounds.
 class OutputStreamPollerImpl : public GraphOutputStream {
  public:
   virtual ~OutputStreamPollerImpl() {}

mediapipe/framework/mediapipe_options.proto

@@ -20,6 +20,9 @@ syntax = "proto2";
 
 package mediapipe;
 
+option java_package = "com.google.mediapipe.proto";
+option java_outer_classname = "MediaPipeOptionsProto";
+
 // Options used by a MediaPipe object.
 message MediaPipeOptions {
   extensions 20000 to max;

mediapipe/framework/output_stream_shard.cc

@@ -101,8 +101,8 @@ Status OutputStreamShard::AddPacketInternal(T&& packet) {
   }
 
   if (packet.IsEmpty()) {
-    return mediapipe::InvalidArgumentErrorBuilder(MEDIAPIPE_LOC)
-           << "Empty packet sent to stream \"" << Name() << "\".";
+    SetNextTimestampBound(packet.Timestamp().NextAllowedInStream());
+    return absl::OkStatus();
   }
 
   const Timestamp timestamp = packet.Timestamp();

mediapipe/framework/tool/BUILD

@@ -20,6 +20,7 @@ load(
     "mediapipe_binary_graph",
 )
 load("//mediapipe/framework:mediapipe_cc_test.bzl", "mediapipe_cc_test")
+load("@bazel_skylib//:bzl_library.bzl", "bzl_library")
 
 licenses(["notice"])
 
@@ -29,6 +30,30 @@ exports_files([
     "simple_subgraph_template.cc",
 ])
 
+bzl_library(
+    name = "mediapipe_graph_bzl",
+    srcs = [
+        "mediapipe_graph.bzl",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":build_defs_bzl",
+        "//mediapipe/framework:encode_binary_proto",
+        "//mediapipe/framework:transitive_protos_bzl",
+        "//mediapipe/framework/deps:expand_template_bzl",
+    ],
+)
+
+bzl_library(
+    name = "build_defs_bzl",
+    srcs = [
+        "build_defs.bzl",
+    ],
+    visibility = [
+        "//mediapipe/framework:__subpackages__",
+    ],
+)
+
 cc_library(
     name = "text_to_binary_graph",
     srcs = ["text_to_binary_graph.cc"],
@@ -744,5 +769,7 @@ cc_test(
 
 exports_files(
     ["build_defs.bzl"],
-    visibility = ["//mediapipe/framework:__subpackages__"],
+    visibility = [
+        "//mediapipe/framework:__subpackages__",
+    ],
 )

mediapipe/framework/tool/sink.cc

@@ -19,6 +19,7 @@
 #include "mediapipe/framework/tool/sink.h"
 
 #include <memory>
+#include <utility>
 #include <vector>
 
 #include "absl/strings/str_cat.h"
@@ -168,8 +169,19 @@ void AddMultiStreamCallback(
     std::function<void(const std::vector<Packet>&)> callback,
     CalculatorGraphConfig* config,
     std::pair<std::string, Packet>* side_packet) {
+  std::map<std::string, Packet> side_packets;
+  AddMultiStreamCallback(streams, callback, config, &side_packets,
+                         /*observe_timestamp_bounds=*/false);
+  *side_packet = *side_packets.begin();
+}
+
+void AddMultiStreamCallback(
+    const std::vector<std::string>& streams,
+    std::function<void(const std::vector<Packet>&)> callback,
+    CalculatorGraphConfig* config, std::map<std::string, Packet>* side_packets,
+    bool observe_timestamp_bounds) {
   CHECK(config);
-  CHECK(side_packet);
+  CHECK(side_packets);
   CalculatorGraphConfig::Node* sink_node = config->add_node();
   const std::string name = GetUnusedNodeName(
       *config, absl::StrCat("multi_callback_", absl::StrJoin(streams, "_")));
@@ -179,15 +191,23 @@ void AddMultiStreamCallback(
     sink_node->add_input_stream(stream_name);
   }
 
+  if (observe_timestamp_bounds) {
+    const std::string observe_ts_bounds_packet_name = GetUnusedSidePacketName(
+        *config, absl::StrCat(name, "_observe_ts_bounds"));
+    sink_node->add_input_side_packet(absl::StrCat(
+        "OBSERVE_TIMESTAMP_BOUNDS:", observe_ts_bounds_packet_name));
+    InsertIfNotPresent(side_packets, observe_ts_bounds_packet_name,
+                       MakePacket<bool>(true));
+  }
   const std::string input_side_packet_name =
       GetUnusedSidePacketName(*config, absl::StrCat(name, "_callback"));
-  side_packet->first = input_side_packet_name;
   sink_node->add_input_side_packet(
       absl::StrCat("VECTOR_CALLBACK:", input_side_packet_name));
 
-  side_packet->second =
+  InsertIfNotPresent(
+      side_packets, input_side_packet_name,
       MakePacket<std::function<void(const std::vector<Packet>&)>>(
-          std::move(callback));
+          std::move(callback)));
 }
 
 void AddCallbackWithHeaderCalculator(const std::string& stream_name,
@@ -240,6 +260,10 @@ absl::Status CallbackCalculator::GetContract(CalculatorContract* cc) {
     return mediapipe::InvalidArgumentErrorBuilder(MEDIAPIPE_LOC)
            << "InputSidePackets must use tags.";
   }
+  if (cc->InputSidePackets().HasTag("OBSERVE_TIMESTAMP_BOUNDS")) {
+    cc->InputSidePackets().Tag("OBSERVE_TIMESTAMP_BOUNDS").Set<bool>();
+    cc->SetProcessTimestampBounds(true);
+  }
 
   int count = allow_multiple_streams ? cc->Inputs().NumEntries("") : 1;
   for (int i = 0; i < count; ++i) {
@@ -266,6 +290,12 @@ absl::Status CallbackCalculator::Open(CalculatorContext* cc) {
     return mediapipe::InvalidArgumentErrorBuilder(MEDIAPIPE_LOC)
            << "missing callback.";
   }
+  if (cc->InputSidePackets().HasTag("OBSERVE_TIMESTAMP_BOUNDS") &&
+      !cc->InputSidePackets().Tag("OBSERVE_TIMESTAMP_BOUNDS").Get<bool>()) {
+    return mediapipe::InvalidArgumentErrorBuilder(MEDIAPIPE_LOC)
+           << "The value of the OBSERVE_TIMESTAMP_BOUNDS input side packet "
+              "must be set to true";
+  }
   return absl::OkStatus();
 }
 

mediapipe/framework/tool/sink.h

@@ -115,6 +115,12 @@ void AddMultiStreamCallback(
     std::function<void(const std::vector<Packet>&)> callback,
     CalculatorGraphConfig* config, std::pair<std::string, Packet>* side_packet);
 
+void AddMultiStreamCallback(
+    const std::vector<std::string>& streams,
+    std::function<void(const std::vector<Packet>&)> callback,
+    CalculatorGraphConfig* config, std::map<std::string, Packet>* side_packets,
+    bool observe_timestamp_bounds = false);
+
 // Add a CallbackWithHeaderCalculator to intercept packets sent on
 // stream stream_name, and the header packet on stream stream_header.
 // The input side packet with the produced name callback_side_packet_name

mediapipe/framework/tool/sink_test.cc

@@ -146,5 +146,63 @@ TEST(CallbackTest, TestAddMultiStreamCallback) {
   EXPECT_THAT(sums, testing::ElementsAre(15, 7, 9));
 }
 
+class TimestampBoundTestCalculator : public CalculatorBase {
+ public:
+  static absl::Status GetContract(CalculatorContract* cc) {
+    cc->Outputs().Index(0).Set<int>();
+    cc->Outputs().Index(1).Set<int>();
+    return absl::OkStatus();
+  }
+  absl::Status Open(CalculatorContext* cc) final { return absl::OkStatus(); }
+  absl::Status Process(CalculatorContext* cc) final {
+    if (count_ % 5 == 0) {
+      cc->Outputs().Index(0).SetNextTimestampBound(Timestamp(count_ + 1));
+      cc->Outputs().Index(1).SetNextTimestampBound(Timestamp(count_ + 1));
+    }
+    ++count_;
+    if (count_ == 13) {
+      return tool::StatusStop();
+    }
+    return absl::OkStatus();
+  }
+
+ private:
+  int count_ = 1;
+};
+REGISTER_CALCULATOR(TimestampBoundTestCalculator);
+
+TEST(CallbackTest, TestAddMultiStreamCallbackWithTimestampNotification) {
+  std::string config_str = R"(
+            node {
+              calculator: "TimestampBoundTestCalculator"
+              output_stream: "foo"
+              output_stream: "bar"
+            }
+          )";
+  CalculatorGraphConfig graph_config =
+      mediapipe::ParseTextProtoOrDie<CalculatorGraphConfig>(config_str);
+
+  std::vector<int> sums;
+
+  std::map<std::string, Packet> side_packets;
+  tool::AddMultiStreamCallback(
+      {"foo", "bar"},
+      [&sums](const std::vector<Packet>& packets) {
+        Packet foo_p = packets[0];
+        Packet bar_p = packets[1];
+        ASSERT_TRUE(foo_p.IsEmpty() && bar_p.IsEmpty());
+        int foo = foo_p.Timestamp().Value();
+        int bar = bar_p.Timestamp().Value();
+        sums.push_back(foo + bar);
+      },
+      &graph_config, &side_packets, true);
+
+  CalculatorGraph graph(graph_config);
+  MP_ASSERT_OK(graph.StartRun(side_packets));
+  MP_ASSERT_OK(graph.WaitUntilDone());
+
+  EXPECT_THAT(sums, testing::ElementsAre(10, 20));
+}
+
 }  // namespace
 }  // namespace mediapipe