Add support for callback file

Change-Id: Ieb116bf3f645a35601f1182ed139c59ddaab8ad8

packager/file/callback_file.cc

@@ -0,0 +1,68 @@
+// Copyright 2017 Google Inc. All rights reserved.
+//
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file or at
+// https://developers.google.com/open-source/licenses/bsd
+
+#include "packager/file/callback_file.h"
+
+#include "packager/base/logging.h"
+
+namespace shaka {
+
+CallbackFile::CallbackFile(const char* file_name, const char* mode)
+    : File(file_name), file_mode_(mode) {}
+
+CallbackFile::~CallbackFile() {}
+
+bool CallbackFile::Close() {
+  delete this;
+  return true;
+}
+
+int64_t CallbackFile::Read(void* buffer, uint64_t length) {
+  if (!callback_params_->read_func) {
+    LOG(ERROR) << "Read function not defined.";
+    return -1;
+  }
+  return callback_params_->read_func(name_, buffer, length);
+}
+
+int64_t CallbackFile::Write(const void* buffer, uint64_t length) {
+  if (!callback_params_->write_func) {
+    LOG(ERROR) << "Write function not defined.";
+    return -1;
+  }
+  return callback_params_->write_func(name_, buffer, length);
+}
+
+int64_t CallbackFile::Size() {
+  LOG(INFO) << "CallbackFile does not support Size().";
+  return -1;
+}
+
+bool CallbackFile::Flush() {
+  // Do nothing on Flush.
+  return true;
+}
+
+bool CallbackFile::Seek(uint64_t position) {
+  VLOG(1) << "CallbackFile does not support Seek().";
+  return false;
+}
+
+bool CallbackFile::Tell(uint64_t* position) {
+  VLOG(1) << "CallbackFile does not support Tell().";
+  return false;
+}
+
+bool CallbackFile::Open() {
+  if (file_mode_ != "r" && file_mode_ != "w" && file_mode_ != "rb" &&
+      file_mode_ != "wb") {
+    LOG(ERROR) << "CallbackFile does not support file mode " << file_mode_;
+    return false;
+  }
+  return ParseCallbackFileName(file_name(), &callback_params_, &name_);
+}
+
+}  // namespace shaka

packager/file/callback_file.h

@@ -0,0 +1,47 @@
+// Copyright 2017 Google Inc. All rights reserved.
+//
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file or at
+// https://developers.google.com/open-source/licenses/bsd
+
+#include "packager/file/file.h"
+
+namespace shaka {
+
+/// Implements CallbackFile, which delegates read/write calls to the callback
+/// functions set through the file name.
+class CallbackFile : public File {
+ public:
+  /// @param file_name is the callback file name, which should have callback
+  ///        address encoded. Note that the file type prefix should be stripped
+  ///        off already.
+  /// @param mode C string containing a file access mode, refer to fopen for
+  ///        the available modes.
+  CallbackFile(const char* file_name, const char* mode);
+
+  /// @name File implementation overrides.
+  /// @{
+  bool Close() override;
+  int64_t Read(void* buffer, uint64_t length) override;
+  int64_t Write(const void* buffer, uint64_t length) override;
+  int64_t Size() override;
+  bool Flush() override;
+  bool Seek(uint64_t position) override;
+  bool Tell(uint64_t* position) override;
+  /// @}
+
+ protected:
+  ~CallbackFile() override;
+
+  bool Open() override;
+
+ private:
+  CallbackFile(const CallbackFile&) = delete;
+  CallbackFile& operator=(const CallbackFile&) = delete;
+
+  const BufferCallbackParams* callback_params_ = nullptr;
+  std::string name_;
+  std::string file_mode_;
+};
+
+}  // namespace shaka

packager/file/callback_file_unittest.cc

@@ -0,0 +1,166 @@
+// Copyright 2017 Google Inc. All rights reserved.
+//
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file or at
+// https://developers.google.com/open-source/licenses/bsd
+
+#include "packager/file/callback_file.h"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "packager/file/file.h"
+#include "packager/file/file_closer.h"
+
+using testing::_;
+using testing::Eq;
+using testing::Invoke;
+using testing::MockFunction;
+using testing::Return;
+using testing::StrEq;
+using testing::WithArgs;
+
+namespace shaka {
+namespace {
+
+const uint8_t kBuffer[] = {1, 2, 3, 4, 5, 6, 7, 8};
+const size_t kBufferSize = sizeof(kBuffer);
+const size_t kSizeLessThanBufferSize = kBufferSize - 2;
+const size_t kSizeLargerThanBufferSize = kBufferSize + 2;
+const char kBufferLabel[] = "some name";
+const int kFileError = -10;
+
+}  // namespace
+
+TEST(CallbackFileTest, ReadSatisfied) {
+  MockFunction<int64_t(const std::string& name, void* buffer, uint64_t length)>
+      mock_read_func;
+  BufferCallbackParams callback_params;
+  callback_params.read_func = mock_read_func.AsStdFunction();
+
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  const size_t size = kSizeLessThanBufferSize;
+
+  EXPECT_CALL(mock_read_func, Call(StrEq(kBufferLabel), _, size))
+      .WillOnce(WithArgs<1, 2>(Invoke([](void* buffer, uint64_t size) {
+        size_t size_to_copy = std::min(static_cast<size_t>(size), kBufferSize);
+        memcpy(buffer, kBuffer, size_to_copy);
+        return size_to_copy;
+      })));
+
+  std::unique_ptr<File, FileCloser> reader(File::Open(file_name.c_str(), "r"));
+  ASSERT_TRUE(reader);
+  uint8_t read_buffer[size];
+  ASSERT_EQ(static_cast<int64_t>(size), reader->Read(read_buffer, size));
+  EXPECT_EQ(0, memcmp(kBuffer, read_buffer, size));
+}
+
+TEST(CallbackFileTest, ReadNotSatisfied) {
+  MockFunction<int64_t(const std::string& name, void* buffer, uint64_t length)>
+      mock_read_func;
+  BufferCallbackParams callback_params;
+  callback_params.read_func = mock_read_func.AsStdFunction();
+
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  const size_t size = kSizeLargerThanBufferSize;
+
+  EXPECT_CALL(mock_read_func, Call(StrEq(kBufferLabel), _, size))
+      .WillOnce(WithArgs<1, 2>(Invoke([](void* buffer, uint64_t size) {
+        size_t size_to_copy = std::min(static_cast<size_t>(size), kBufferSize);
+        memcpy(buffer, kBuffer, size_to_copy);
+        return size_to_copy;
+      })));
+
+  std::unique_ptr<File, FileCloser> reader(File::Open(file_name.c_str(), "r"));
+  ASSERT_TRUE(reader);
+  uint8_t read_buffer[size];
+  ASSERT_EQ(static_cast<int64_t>(kBufferSize), reader->Read(read_buffer, size));
+  EXPECT_EQ(0, memcmp(kBuffer, read_buffer, kBufferSize));
+}
+
+TEST(CallbackFileTest, ReadFailed) {
+  MockFunction<int64_t(const std::string& name, void* buffer, uint64_t length)>
+      mock_read_func;
+  BufferCallbackParams callback_params;
+  callback_params.read_func = mock_read_func.AsStdFunction();
+
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  EXPECT_CALL(mock_read_func, Call(StrEq(kBufferLabel), _, _))
+      .WillOnce(WithArgs<1, 2>(
+          Invoke([](void* buffer, uint64_t size) { return kFileError; })));
+
+  std::unique_ptr<File, FileCloser> reader(File::Open(file_name.c_str(), "r"));
+  ASSERT_TRUE(reader);
+  uint8_t read_buffer[kBufferSize];
+  ASSERT_EQ(kFileError, reader->Read(read_buffer, kBufferSize));
+}
+
+TEST(CallbackFileTest, ReadFunctionNotDefined) {
+  BufferCallbackParams callback_params;
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  std::unique_ptr<File, FileCloser> reader(File::Open(file_name.c_str(), "r"));
+  ASSERT_TRUE(reader);
+  uint8_t read_buffer[kBufferSize];
+  ASSERT_EQ(-1, reader->Read(read_buffer, kBufferSize));
+}
+
+TEST(CallbackFileTest, WriteSatisfied) {
+  MockFunction<int64_t(const std::string& name, const void* buffer,
+                       uint64_t length)>
+      mock_write_func;
+  BufferCallbackParams callback_params;
+  callback_params.write_func = mock_write_func.AsStdFunction();
+
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  EXPECT_CALL(mock_write_func,
+              Call(StrEq(kBufferLabel), Eq(kBuffer), kBufferSize))
+      .WillOnce(Return(kBufferSize));
+
+  std::unique_ptr<File, FileCloser> writer(File::Open(file_name.c_str(), "w"));
+  ASSERT_TRUE(writer);
+  ASSERT_EQ(static_cast<int64_t>(kBufferSize),
+            writer->Write(kBuffer, kBufferSize));
+}
+
+TEST(CallbackFileTest, WriteFailed) {
+  MockFunction<int64_t(const std::string& name, const void* buffer,
+                       uint64_t length)>
+      mock_write_func;
+  BufferCallbackParams callback_params;
+  callback_params.write_func = mock_write_func.AsStdFunction();
+
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  EXPECT_CALL(mock_write_func,
+              Call(StrEq(kBufferLabel), Eq(kBuffer), kBufferSize))
+      .WillOnce(Return(kFileError));
+
+  std::unique_ptr<File, FileCloser> writer(File::Open(file_name.c_str(), "w"));
+  ASSERT_TRUE(writer);
+  ASSERT_EQ(kFileError, writer->Write(kBuffer, kBufferSize));
+}
+
+TEST(CallbackFileTest, WriteFunctionNotDefined) {
+  BufferCallbackParams callback_params;
+  std::string file_name =
+      File::MakeCallbackFileName(callback_params, kBufferLabel);
+
+  std::unique_ptr<File, FileCloser> writer(File::Open(file_name.c_str(), "w"));
+  ASSERT_TRUE(writer);
+  ASSERT_EQ(-1, writer->Write(kBuffer, kBufferSize));
+}
+
+}  // namespace shaka

packager/file/file.cc

@@ -7,11 +7,15 @@
 #include "packager/file/file.h"
 
 #include <gflags/gflags.h>
+#include <inttypes.h>
 #include <algorithm>
 #include <memory>
 #include "packager/base/files/file_util.h"
 #include "packager/base/logging.h"
+#include "packager/base/strings/string_number_conversions.h"
 #include "packager/base/strings/string_piece.h"
+#include "packager/base/strings/stringprintf.h"
+#include "packager/file/callback_file.h"
 #include "packager/file/file_util.h"
 #include "packager/file/local_file.h"
 #include "packager/file/memory_file.h"
@@ -33,9 +37,10 @@ DEFINE_uint64(io_block_size,
 
 namespace shaka {
 
+const char* kCallbackFilePrefix = "callback://";
 const char* kLocalFilePrefix = "file://";
-const char* kUdpFilePrefix = "udp://";
 const char* kMemoryFilePrefix = "memory://";
+const char* kUdpFilePrefix = "udp://";
 
 namespace {
 
@@ -46,12 +51,15 @@ typedef bool (*FileAtomicWriteFunction)(const char* file_name,
 
 struct FileTypeInfo {
   const char* type;
-  size_t type_length;
   const FileFactoryFunction factory_function;
   const FileDeleteFunction delete_function;
   const FileAtomicWriteFunction atomic_write_function;
 };
 
+File* CreateCallbackFile(const char* file_name, const char* mode) {
+  return new CallbackFile(file_name, mode);
+}
+
 File* CreateLocalFile(const char* file_name, const char* mode) {
   return new LocalFile(file_name, mode);
 }
@@ -99,21 +107,26 @@ bool DeleteMemoryFile(const char* file_name) {
 static const FileTypeInfo kFileTypeInfo[] = {
     {
         kLocalFilePrefix,
-        strlen(kLocalFilePrefix),
         &CreateLocalFile,
         &DeleteLocalFile,
         &WriteLocalFileAtomically,
     },
-    {kUdpFilePrefix, strlen(kUdpFilePrefix), &CreateUdpFile, nullptr, nullptr},
-    {kMemoryFilePrefix, strlen(kMemoryFilePrefix), &CreateMemoryFile,
-     &DeleteMemoryFile, nullptr},
+    {kUdpFilePrefix, &CreateUdpFile, nullptr, nullptr},
+    {kMemoryFilePrefix, &CreateMemoryFile, &DeleteMemoryFile, nullptr},
+    {kCallbackFilePrefix, &CreateCallbackFile, nullptr, nullptr},
 };
 
+base::StringPiece GetFileTypePrefix(base::StringPiece file_name) {
+  size_t pos = file_name.find("://");
+  return (pos == std::string::npos) ? "" : file_name.substr(0, pos + 3);
+}
+
 const FileTypeInfo* GetFileTypeInfo(base::StringPiece file_name,
                                     base::StringPiece* real_file_name) {
+  base::StringPiece file_type_prefix = GetFileTypePrefix(file_name);
   for (const FileTypeInfo& file_type : kFileTypeInfo) {
-    if (strncmp(file_type.type, file_name.data(), file_type.type_length) == 0) {
-      *real_file_name = file_name.substr(file_type.type_length);
+    if (file_type_prefix == file_type.type) {
+      *real_file_name = file_name.substr(file_type_prefix.size());
       return &file_type;
     }
   }
@@ -128,8 +141,10 @@ File* File::Create(const char* file_name, const char* mode) {
   std::unique_ptr<File, FileCloser> internal_file(
       CreateInternalFile(file_name, mode));
 
-  if (!strncmp(file_name, kMemoryFilePrefix, strlen(kMemoryFilePrefix))) {
-    // Disable caching for memory files.
+  base::StringPiece file_type_prefix = GetFileTypePrefix(file_name);
+  if (file_type_prefix == kMemoryFilePrefix ||
+      file_type_prefix == kCallbackFilePrefix) {
+    // Disable caching for memory and callback files.
     return internal_file.release();
   }
 
@@ -328,4 +343,32 @@ int64_t File::CopyFile(File* source, File* destination, int64_t max_copy) {
   return bytes_copied;
 }
 
+std::string File::MakeCallbackFileName(
+    const BufferCallbackParams& callback_params,
+    const std::string& name) {
+  if (name.empty())
+    return "";
+  return base::StringPrintf("%s%" PRIdPTR "/%s", kCallbackFilePrefix,
+                            reinterpret_cast<intptr_t>(&callback_params),
+                            name.c_str());
+}
+
+bool File::ParseCallbackFileName(const std::string& callback_file_name,
+                                 const BufferCallbackParams** callback_params,
+                                 std::string* name) {
+  size_t pos = callback_file_name.find("/");
+  int64_t callback_address = 0;
+  if (pos == std::string::npos ||
+      !base::StringToInt64(callback_file_name.substr(0, pos),
+                           &callback_address)) {
+    LOG(ERROR) << "Expecting CallbackFile with name like "
+                  "'<callback address>/<entity name>', but seeing "
+               << callback_file_name;
+    return false;
+  }
+  *callback_params = reinterpret_cast<BufferCallbackParams*>(callback_address);
+  *name = callback_file_name.substr(pos + 1);
+  return true;
+}
+
 }  // namespace shaka

packager/file/file.gyp

@@ -13,6 +13,8 @@
       'target_name': 'file',
       'type': '<(component)',
       'sources': [
+        'callback_file.cc',
+        'callback_file.h',
         'file.cc',
         'file.h',
         'file_util.cc',
@@ -24,6 +26,7 @@
         'local_file.h',
         'memory_file.cc',
         'memory_file.h',
+        'public/buffer_callback_params.h',
         'threaded_io_file.cc',
         'threaded_io_file.h',
         'udp_file.cc',
@@ -40,6 +43,7 @@
       'target_name': 'file_unittest',
       'type': '<(gtest_target_type)',
       'sources': [
+        'callback_file_unittest.cc',
         'file_unittest.cc',
         'file_util_unittest.cc',
         'io_cache_unittest.cc',
@@ -48,6 +52,7 @@
       ],
       'dependencies': [
         '../media/test/media_test.gyp:run_tests_with_atexit_manager',
+        '../testing/gmock.gyp:gmock',
         '../testing/gtest.gyp:gtest',
         '../third_party/gflags/gflags.gyp:gflags',
         'file',

packager/file/file.h

@@ -12,11 +12,14 @@
 #include <string>
 
 #include "packager/base/macros.h"
+#include "packager/file/public/buffer_callback_params.h"
 
 namespace shaka {
 
+extern const char* kCallbackFilePrefix;
 extern const char* kLocalFilePrefix;
 extern const char* kMemoryFilePrefix;
+extern const char* kUdpFilePrefix;
 const int64_t kWholeFile = -1;
 
 /// Define an abstract file interface.
@@ -86,7 +89,8 @@ class File {
   /// @return true on succcess, false otherwise.
   virtual bool Tell(uint64_t* position) = 0;
 
-  /// @return The file name.
+  /// @return The file name. Note that the file type prefix has been stripped
+  ///         off.
   const std::string& file_name() const { return file_name_; }
 
   // ************************************************************
@@ -138,6 +142,27 @@ class File {
   /// @return Number of bytes written, or a value < 0 on error.
   static int64_t CopyFile(File* source, File* destination, int64_t max_copy);
 
+  /// Generate callback file name.
+  /// NOTE: THE GENERATED NAME IS ONLY VAID WHILE @a callback_params IS VALID.
+  /// @param callback_params references BufferCallbackParams, which will be
+  ///        embedded in the generated callback file name.
+  /// @param name is the name of the buffer, which will be embedded in the
+  ///        generated callback file name.
+  static std::string MakeCallbackFileName(
+      const BufferCallbackParams& callback_params,
+      const std::string& name);
+
+  /// Parse and extract callback params.
+  /// @param callback_file_name is the name of the callback file which contains
+  ///        @a callback_params and @a name.
+  /// @param callback_params points to the parsed BufferCallbackParams pointer.
+  /// @param name points to the parsed name.
+  /// @return true on success, false otherwise.
+  static bool ParseCallbackFileName(
+      const std::string& callback_file_name,
+      const BufferCallbackParams** callback_params,
+      std::string* name);
+
  protected:
   explicit File(const std::string& file_name) : file_name_(file_name) {}
   /// Do *not* call the destructor directly (with the "delete" keyword)
@@ -156,7 +181,9 @@ class File {
 
   static File* CreateInternalFile(const char* file_name, const char* mode);
 
+  // Note that the file type prefix has been stripped off.
   std::string file_name_;
+
   DISALLOW_COPY_AND_ASSIGN(File);
 };
 

packager/file/file_unittest.cc

@@ -300,4 +300,27 @@ TEST_F(LocalFileTest, DISABLED_ReadSeekOutOfBounds) {
   EXPECT_TRUE(file->Close());
 }
 
+TEST(FileTest, MakeCallbackFileName) {
+  const BufferCallbackParams* params =
+      reinterpret_cast<BufferCallbackParams*>(1000);
+  EXPECT_EQ("callback://1000/some name",
+            File::MakeCallbackFileName(*params, "some name"));
+  EXPECT_EQ("", File::MakeCallbackFileName(*params, ""));
+}
+
+TEST(FileTest, ParseCallbackFileName) {
+  const BufferCallbackParams* params = nullptr;
+  std::string name;
+  ASSERT_TRUE(File::ParseCallbackFileName("1000/some name", &params, &name));
+  EXPECT_EQ(1000, reinterpret_cast<int64_t>(params));
+  EXPECT_EQ("some name", name);
+}
+
+TEST(FileTest, ParseCallbackFileNameFailed) {
+  const BufferCallbackParams* params = nullptr;
+  std::string name;
+  ASSERT_FALSE(File::ParseCallbackFileName("1000\\some name", &params, &name));
+  ASSERT_FALSE(File::ParseCallbackFileName("abc/some name", &params, &name));
+}
+
 }  // namespace shaka

packager/file/public/buffer_callback_params.h

@@ -0,0 +1,35 @@
+// Copyright 2017 Google Inc. All rights reserved.
+//
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file or at
+// https://developers.google.com/open-source/licenses/bsd
+
+#ifndef PACKAGER_FILE_PUBLIC_BUFFER_CALLBACK_PARAMS_H_
+#define PACKAGER_FILE_PUBLIC_BUFFER_CALLBACK_PARAMS_H_
+
+#include <functional>
+
+namespace shaka {
+
+/// Buffer callback params.
+struct BufferCallbackParams {
+  /// If this function is specified, packager treats @a StreamDescriptor.input
+  /// as a label and call this function with @a name set to
+  /// @a StreamDescriptor.input.
+  std::function<int64_t(const std::string& name, void* buffer, uint64_t size)>
+      read_func;
+  /// If this function is specified, packager treats the output files specified
+  /// in PackagingParams and StreamDescriptors as labels and calls this function
+  /// with @a name set. This applies to @a
+  /// PackagingParams.MpdParams.mpd_output,
+  /// @a PackagingParams.HlsParams.master_playlist_output, @a
+  /// StreamDescriptor.output, @a StreamDescriptor.segment_template, @a
+  /// StreamDescriptor.hls_playlist_name.
+  std::function<
+      int64_t(const std::string& name, const void* buffer, uint64_t size)>
+      write_func;
+};
+
+}  // namespace shaka
+
+#endif  // PACKAGER_FILE_PUBLIC_BUFFER_CALLBACK_PARAMS_H_

packager/packager.cc

@@ -612,6 +612,7 @@ struct Packager::PackagerInternal {
   std::unique_ptr<MpdNotifier> mpd_notifier;
   std::unique_ptr<hls::HlsNotifier> hls_notifier;
   std::vector<std::unique_ptr<media::Job>> jobs;
+  BufferCallbackParams buffer_callback_params;
 };
 
 Packager::Packager() {}
@@ -651,12 +652,24 @@ Status Packager::Initialize(
       return Status(error::INVALID_ARGUMENT, "Failed to create key source.");
   }
 
-  const MpdParams& mpd_params = packaging_params.mpd_params;
+  // Store callback params to make it available during packaging.
+  internal->buffer_callback_params = packaging_params.buffer_callback_params;
+
+  // Update mpd output and hls output if callback param is specified.
+  MpdParams mpd_params = packaging_params.mpd_params;
+  HlsParams hls_params = packaging_params.hls_params;
+  if (internal->buffer_callback_params.write_func) {
+    mpd_params.mpd_output = File::MakeCallbackFileName(
+        internal->buffer_callback_params, mpd_params.mpd_output);
+    hls_params.master_playlist_output = File::MakeCallbackFileName(
+        internal->buffer_callback_params, hls_params.master_playlist_output);
+  }
+
   if (!mpd_params.mpd_output.empty()) {
     const bool on_demand_dash_profile =
         stream_descriptors.begin()->segment_template.empty();
-    MpdOptions mpd_options = media::GetMpdOptions(on_demand_dash_profile,
-                                                  packaging_params.mpd_params);
+    MpdOptions mpd_options =
+        media::GetMpdOptions(on_demand_dash_profile, mpd_params);
     if (mpd_params.generate_dash_if_iop_compliant_mpd) {
       internal->mpd_notifier.reset(new DashIopMpdNotifier(mpd_options));
     } else {
@@ -669,7 +682,6 @@ Status Packager::Initialize(
     }
   }
 
-  const HlsParams& hls_params = packaging_params.hls_params;
   if (!hls_params.master_playlist_output.empty()) {
     base::FilePath master_playlist_path(
         base::FilePath::FromUTF8Unsafe(hls_params.master_playlist_output));
@@ -683,8 +695,26 @@ Status Packager::Initialize(
   }
 
   media::StreamDescriptorList stream_descriptor_list;
-  for (const StreamDescriptor& descriptor : stream_descriptors)
-    stream_descriptor_list.insert(descriptor);
+  for (const StreamDescriptor& descriptor : stream_descriptors) {
+    if (internal->buffer_callback_params.read_func ||
+        internal->buffer_callback_params.write_func) {
+      StreamDescriptor descriptor_copy = descriptor;
+      if (internal->buffer_callback_params.read_func) {
+        descriptor_copy.input = File::MakeCallbackFileName(
+            internal->buffer_callback_params, descriptor.input);
+      }
+      if (internal->buffer_callback_params.write_func) {
+        descriptor_copy.output = File::MakeCallbackFileName(
+            internal->buffer_callback_params, descriptor.output);
+        descriptor_copy.segment_template = File::MakeCallbackFileName(
+            internal->buffer_callback_params, descriptor.segment_template);
+      }
+      stream_descriptor_list.insert(descriptor_copy);
+    } else {
+      stream_descriptor_list.insert(descriptor);
+    }
+  }
+
   Status status = media::CreateRemuxJobs(
       stream_descriptor_list, packaging_params, &internal->fake_clock,
       internal->encryption_key_source.get(), internal->mpd_notifier.get(),

packager/packager.gyp

@@ -119,6 +119,7 @@
       'dependencies': [
         'base/base.gyp:base',
         'libpackager',
+        'testing/gmock.gyp:gmock',
         'testing/gtest.gyp:gtest',
         'testing/gtest.gyp:gtest_main',
       ],

packager/packager.h

@@ -12,6 +12,7 @@
 #include <vector>
 
 #include "packager/hls/public/hls_params.h"
+#include "packager/file/public/buffer_callback_params.h"
 #include "packager/media/public/chunking_params.h"
 #include "packager/media/public/crypto_params.h"
 #include "packager/media/public/mp4_output_params.h"
@@ -55,6 +56,9 @@ struct PackagingParams {
   EncryptionParams encryption_params;
   DecryptionParams decryption_params;
 
+  /// Buffer callback params.
+  BufferCallbackParams buffer_callback_params;
+
   // Parameters for testing. Do not use in production.
   TestParams test_params;
 };
@@ -63,8 +67,6 @@ struct PackagingParams {
 struct StreamDescriptor {
   /// Input/source media file path or network stream URL. Required.
   std::string input;
-  // TODO(kqyang): Add support for feeding data through read func.
-  // std::function<int64_t(void* buffer, uint64_t length)> read_func;
 
   /// Stream selector, can be `audio`, `video`, `text` or a zero based stream
   /// index. Required.
@@ -75,9 +77,6 @@ struct StreamDescriptor {
   std::string output;
   /// Specifies segment template. Can be empty.
   std::string segment_template;
-  // TODO: Add support for writing data through write func.
-  // std::function<int64_t(const std::string& id, void* buffer, uint64_t
-  // length)> write_func;
 
   /// Optional value which specifies output container format, e.g. "mp4". If not
   /// specified, will detect from output / segment template name.

packager/packager_test.cc

@@ -4,6 +4,7 @@
 // license that can be found in the LICENSE file or at
 // https://developers.google.com/open-source/licenses/bsd
 
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include "packager/base/files/file_util.h"
@@ -12,6 +13,14 @@
 #include "packager/base/strings/string_number_conversions.h"
 #include "packager/packager.h"
 
+using testing::_;
+using testing::Invoke;
+using testing::MockFunction;
+using testing::Return;
+using testing::ReturnArg;
+using testing::StrEq;
+using testing::WithArgs;
+
 namespace shaka {
 namespace {
 
@@ -154,6 +163,69 @@ TEST_F(PackagerTest, SegmentNotAlignedButSubsegmentAligned) {
   ASSERT_EQ(error::INVALID_ARGUMENT, status.error_code());
 }
 
+TEST_F(PackagerTest, WriteOutputToBuffer) {
+  auto packaging_params = SetupPackagingParams();
+
+  MockFunction<int64_t(const std::string& name, const void* buffer,
+                       uint64_t length)>
+      mock_write_func;
+  packaging_params.buffer_callback_params.write_func =
+      mock_write_func.AsStdFunction();
+  EXPECT_CALL(mock_write_func, Call(StrEq(GetFullPath(kOutputVideo)), _, _))
+      .WillRepeatedly(ReturnArg<2>());
+  EXPECT_CALL(mock_write_func, Call(StrEq(GetFullPath(kOutputAudio)), _, _))
+      .WillRepeatedly(ReturnArg<2>());
+  EXPECT_CALL(mock_write_func, Call(StrEq(GetFullPath(kOutputMpd)), _, _))
+      .WillRepeatedly(ReturnArg<2>());
+
+  Packager packager;
+  ASSERT_EQ(Status::OK,
+            packager.Initialize(packaging_params, SetupStreamDescriptors()));
+  ASSERT_EQ(Status::OK, packager.Run());
+}
+
+TEST_F(PackagerTest, ReadFromBuffer) {
+  auto packaging_params = SetupPackagingParams();
+
+  MockFunction<int64_t(const std::string& name, void* buffer, uint64_t length)>
+      mock_read_func;
+  packaging_params.buffer_callback_params.read_func =
+      mock_read_func.AsStdFunction();
+
+  const std::string file_name = GetTestDataFilePath(kTestFile);
+  FILE* file_ptr =
+      base::OpenFile(base::FilePath::FromUTF8Unsafe(file_name), "rb");
+  ASSERT_TRUE(file_ptr);
+  EXPECT_CALL(mock_read_func, Call(StrEq(file_name), _, _))
+      .WillRepeatedly(
+          WithArgs<1, 2>(Invoke([file_ptr](void* buffer, uint64_t size) {
+            return fread(buffer, sizeof(char), size, file_ptr);
+          })));
+
+  Packager packager;
+  ASSERT_EQ(Status::OK,
+            packager.Initialize(packaging_params, SetupStreamDescriptors()));
+  ASSERT_EQ(Status::OK, packager.Run());
+
+  base::CloseFile(file_ptr);
+}
+
+TEST_F(PackagerTest, ReadFromBufferFailed) {
+  auto packaging_params = SetupPackagingParams();
+
+  MockFunction<int64_t(const std::string& name, void* buffer, uint64_t length)>
+      mock_read_func;
+  packaging_params.buffer_callback_params.read_func =
+      mock_read_func.AsStdFunction();
+
+  EXPECT_CALL(mock_read_func, Call(_, _, _)).WillOnce(Return(-1));
+
+  Packager packager;
+  ASSERT_EQ(Status::OK,
+            packager.Initialize(packaging_params, SetupStreamDescriptors()));
+  ASSERT_EQ(error::FILE_FAILURE, packager.Run().error_code());
+}
+
 // TODO(kqyang): Add more tests.
 
 }  // namespace shaka