(feat): Windows support

.gitignore

@@ -11,4 +11,5 @@ shaka_streamer.egg-info/
 .idea/
 venv/
 dev/
-.vscode/
+.vscode/
+packager.exe

run_end_to_end_tests.py

@@ -125,7 +125,9 @@ def hlsStreamsReady(master_playlist_path):
     return False
 
   for playlist_path in playlist_list:
-    if playlist_path == master_playlist_path:
+    # Use os.path.samefile method instead of the == operator because
+    # this might be a windows machine.
+    if os.path.samefile(playlist_path, master_playlist_path):
       # Skip the master playlist
       continue
 
@@ -269,7 +271,8 @@ def main():
     return 1
 
   # Install test dependencies.
-  subprocess.check_call(['npm', 'install'])
+  npm_command = ' '.join(['npm', 'install'])
+  subprocess.check_call(npm_command, shell=True)
 
   # Fetch streams used in tests.
   if not os.path.exists(TEST_DIR):
@@ -289,24 +292,23 @@ def main():
 
   for i in range(trials):
     # Start up karma.
-    karma_args = [
+    karma_command = ' '.join([
+        'node',
         'node_modules/karma/bin/karma',
         'start',
         'tests/karma.conf.js',
         # DRM currently is not compatible with headless, so it's run in Chrome.
         # Linux: If you want to run tests as "headless", wrap it with "xvfb-run -a".
         '--browsers', 'Chrome',
         '--single-run',
-      ]
+      ])
 
     if args.reporters:
       converted_string = ','.join(args.reporters)
-      karma_args += [
-          '--reporters',
-          converted_string,
-      ]
+      karma_command += ' --reporters {}'.format(converted_string)
+
     # If the exit code was not 0, the tests in karma failed or crashed.
-    if subprocess.call(karma_args) != 0:
+    if subprocess.call(karma_command, shell=True) != 0:
       fails += 1
 
   print('\n\nNumber of failures:', fails, '\nNumber of trials:', trials)

setup.py

@@ -13,13 +13,18 @@
 # limitations under the License.
 
 import base64
+import os
 import setuptools
 
 import streamer
 
 with open('README.md', 'r') as f:
   long_description = f.read()
 
+install_prerequisites = ['PyYAML']
+if os.name == 'nt':
+  install_prerequisites.append('pypiwin32')
+
 setuptools.setup(
   name='shaka-streamer',
   version=streamer.__version__,
@@ -29,9 +34,7 @@
   long_description_content_type='text/markdown',
   url='https://github.com/google/shaka-streamer',
   packages=setuptools.find_packages(),
-  install_requires=[
-      'PyYAML',
-  ],
+  install_requires=install_prerequisites,
   scripts=['shaka-streamer'],
   classifiers=[
       'Programming Language :: Python :: 3',

streamer/controller_node.py

@@ -41,6 +41,7 @@
 from streamer.pipeline_configuration import PipelineConfig, StreamingMode
 from streamer.transcoder_node import TranscoderNode
 from streamer.periodconcat_node import PeriodConcatNode
+from streamer.winfifo import WinFIFO
 
 
 class ControllerNode(object):
@@ -79,21 +80,23 @@ def _create_pipe(self, suffix = '') -> str:
     Returns:
       The path to the named pipe, as a string.
     """
-
-    # TODO(#8): mkfifo only works on Unix.  We would need a special case for a
-    # Windows port some day.
-
-    if not hasattr(os, 'mkfifo'):
-      raise RuntimeError('Platform not supported due to lack of mkfifo')
-
     # Since the tempfile module creates actual files, use uuid to generate a
     # filename, then call mkfifo to create the named pipe.
     unique_name = str(uuid.uuid4()) + suffix
-    path = os.path.join(self._temp_dir, unique_name)
-
-    readable_by_owner_only = 0o600  # Unix permission bits
-    os.mkfifo(path, mode=readable_by_owner_only)
-
+
+    # For POSIX systems.
+    if os.name == 'posix':
+      path = os.path.join(self._temp_dir, unique_name)
+      readable_by_owner_only = 0o600  # Unix permission bits
+      os.mkfifo(path, mode=readable_by_owner_only) # type: ignore
+
+    # New Technology, aka WindowsNT.
+    elif os.name == 'nt':
+      path = '-nt-shaka-' + unique_name
+      WinFIFO(path).start()
+    else:
+      raise RuntimeError('Platform not supported.')
+
     return path
 
   def start(self, output_dir: str,
@@ -209,9 +212,14 @@ def _append_nodes_for_inputs_list(self, inputs: List[Input],
       # read from that pipe for this input.
       if input.input_type == InputType.EXTERNAL_COMMAND:
         command_output = self._create_pipe()
+        writer_command_output = command_output
+        reader_command_output = command_output
+        if os.name == 'nt':
+          writer_command_output = WinFIFO.WRITER_PREFIX + command_output
+          reader_command_output = WinFIFO.READER_PREFIX + command_output
         self._nodes.append(ExternalCommandNode(
-            input.name, command_output))
-        input.set_pipe(command_output)
+            input.name, writer_command_output))
+        input.set_pipe(reader_command_output)
 
       if input.media_type == MediaType.AUDIO:
         for audio_codec in self._pipeline_config.audio_codecs:

streamer/output_stream.py

@@ -14,8 +14,11 @@
 
 """Contains information about each output stream."""
 
+import os
+
 from streamer.bitrate_configuration import AudioCodec, AudioChannelLayout, VideoCodec, VideoResolution
 from streamer.input_configuration import Input, MediaType
+from streamer.winfifo import WinFIFO
 from typing import Dict, Optional, Union
 
 
@@ -29,7 +32,14 @@ def __init__(self,
                codec: Union[AudioCodec, VideoCodec, None]) -> None:
     self.type: MediaType = type
     # If "pipe" is None, then this will not be transcoded.
-    self.pipe: Optional[str] = pipe
+    self.read_pipe: Optional[str] = pipe
+    self.writ_pipe: Optional[str] = pipe
+    # On posix systems, the read and write pipes will be the same.
+    # But on Windows, we will have different read and write pipe names.
+    if os.name == 'nt' and self.read_pipe and self.writ_pipe:
+      # Prefix the pipe names on Windows platforms.
+      self.read_pipe = WinFIFO.READER_PREFIX + self.read_pipe
+      self.writ_pipe = WinFIFO.WRITER_PREFIX + self.writ_pipe
     self.input: Input = input
     self.codec: Union[AudioCodec, VideoCodec, None] = codec
     self._features: Dict[str, str] = {}

streamer/packager_node.py

@@ -120,10 +120,11 @@ def start(self) -> None:
         stdout=stdout)
 
   def _setup_stream(self, stream: OutputStream) -> str:
+
     dict = {
         # If pipe is None, this wasn't transcoded, so we take the input path
         # directly.
-        'in': stream.pipe or stream.input.name,
+        'in': stream.read_pipe or stream.input.name,
         'stream': stream.type.value,
     }
 

streamer/transcoder_node.py

@@ -123,7 +123,7 @@ def start(self) -> None:
         if output_stream.input != input:
           # Skip outputs that don't match this exact input object.
           continue
-        if output_stream.pipe is None:
+        if output_stream.writ_pipe is None:
           # This input won't be transcoded.  This is common for VTT text input.
           continue
 
@@ -140,8 +140,7 @@ def start(self) -> None:
           assert(isinstance(output_stream, TextOutputStream))
           args += self._encode_text(output_stream, input)
 
-        # The output pipe.
-        args += [output_stream.pipe]
+        args += [output_stream.writ_pipe]
 
     env = {}
     if self._pipeline_config.debug_logs:

streamer/winfifo.py

@@ -0,0 +1,80 @@
+# Copyright 2019 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from threading import Thread
+import win32pipe, win32file, pywintypes # type: ignore
+
+class WinFIFO(Thread):
+  """A threaded class that serves as a FIFO pipe that transfers data from a
+  writer to a reader process on Windows.
+
+  It is a replacement for os.mkfifo on POSIX systems."""
+
+  READER_PREFIX = r'\\.\pipe\R'
+  WRITER_PREFIX = r'\\.\pipe\W'
+
+  def __init__(self, pipe_name: str, buf_size = 64 * 1024):
+    """Initializes a thread and creates two named pipes on the system."""
+
+    super().__init__(daemon=True)
+    self.pipe_name = pipe_name
+    self.BUF_SIZE = buf_size
+
+    # The read pipe is connected to a writer process.
+    self.read_side = win32pipe.CreateNamedPipe(
+    WinFIFO.WRITER_PREFIX + self.pipe_name,
+    win32pipe.PIPE_ACCESS_INBOUND,
+    win32pipe.PIPE_WAIT | win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE,
+    1,
+    self.BUF_SIZE,
+    self.BUF_SIZE,
+    0,
+    None)
+
+    # The write pipe is connected to a reader process.
+    self.writ_side = win32pipe.CreateNamedPipe(
+    WinFIFO.READER_PREFIX + self.pipe_name,
+    win32pipe.PIPE_ACCESS_OUTBOUND,
+    win32pipe.PIPE_WAIT | win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE,
+    1,
+    self.BUF_SIZE,
+    self.BUF_SIZE,
+    0,
+    None)
+
+  def run(self):
+    """This method serves as a server that connects a writer client
+    to a reader client.
+
+    This methods will run as a thread."""
+
+    try:
+      # Connect to both ends of the pipe before starting the transfer.
+      # This funciton is blocking. If no process is connected yet, it will wait
+      # indefinitely.
+      win32pipe.ConnectNamedPipe(self.read_side)
+      win32pipe.ConnectNamedPipe(self.writ_side)
+      while True:
+        # Writer -> read_side -> writ_side -> Reader
+        _, data = win32file.ReadFile(self.read_side, self.BUF_SIZE)
+        win32file.WriteFile(self.writ_side, data)
+    except Exception as ex:
+      # Remove the pipes from the system.
+      win32file.CloseHandle(self.read_side)
+      win32file.CloseHandle(self.writ_side)
+      # If the error was due to one of the processes shutting down, just exit normally.
+      if isinstance(ex, pywintypes.error) and (ex.args[0] == 109 or ex.args[0] == 232):
+        return 0
+      # Otherwise, raise that error.
+      raise ex