Repository: streamlink/streamlink
Pull request #3710 (2021-04-28 13:01:32)

plugins.nicolive: fix nicolive broken with proxy
nico live run failed with error:

#> streamlink --http-proxy=127.0.0.1:8080 -o ./out.ts "https://live.nicovideo.jp/watch/lv********" test

[plugins.nicolive][warning] setsockopt() takes exactly 3 arguments (1 given)
[plugins.nicolive][warning] wss api disconnected.
[plugins.nicolive][warning] Attempting to reconnect in 5 secs...

fix: websocket threading should init with `kwargs`, not `args`.


Patch:
diff --git a/docs/install.rst b/docs/install.rst
index 314434bc..e1188aa0 100644
--- a/docs/install.rst
+++ b/docs/install.rst
@@ -34,7 +34,7 @@ Python pip                           See the `PyPI package and source code`_ sec
 
 .. _Chocolatey: https://chocolatey.org/packages/streamlink
 
-.. _Windows Package Manager: https://github.com/microsoft/winget-pkgs/tree/master/manifests/Streamlink/Streamlink
+.. _Windows Package Manager: https://github.com/microsoft/winget-pkgs/tree/master/manifests/s/Streamlink/Streamlink
 
 .. _Installing Chocolatey packages: https://chocolatey.org
 
diff --git a/src/streamlink/plugins/nicolive.py b/src/streamlink/plugins/nicolive.py
index 5608eea0..4f6873fc 100644
--- a/src/streamlink/plugins/nicolive.py
+++ b/src/streamlink/plugins/nicolive.py
@@ -197,7 +197,7 @@ class NicoLive(Plugin):
             on_error=on_error)
         self.ws_worker_thread = threading.Thread(
             target=self._ws.run_forever,
-            args=proxy_options)
+            kwargs=proxy_options)
         self.ws_worker_thread.daemon = True
         self.ws_worker_thread.start()
 
diff --git a/src/streamlink/stream/ffmpegmux.py b/src/streamlink/stream/ffmpegmux.py
index aa319442..135effc9 100644
--- a/src/streamlink/stream/ffmpegmux.py
+++ b/src/streamlink/stream/ffmpegmux.py
@@ -1,6 +1,4 @@
 import logging
-import os
-import random
 import subprocess
 import sys
 import threading
@@ -62,7 +60,7 @@ class FFMPEGMuxer(StreamIO):
     @staticmethod
     def copy_to_pipe(self, stream, pipe):
         log.debug("Starting copy to pipe: {0}".format(pipe.path))
-        pipe.open("wb")
+        pipe.open()
         while not stream.closed:
             try:
                 data = stream.read(8192)
@@ -87,7 +85,7 @@ class FFMPEGMuxer(StreamIO):
         self.process = None
         self.streams = streams
 
-        self.pipes = [NamedPipe("ffmpeg-{0}-{1}".format(os.getpid(), random.randint(0, 1000))) for _ in self.streams]
+        self.pipes = [NamedPipe() for _ in self.streams]
         self.pipe_threads = [threading.Thread(target=self.copy_to_pipe, args=(self, stream, np))
                              for stream, np in
                              zip(self.streams, self.pipes)]
@@ -103,7 +101,7 @@ class FFMPEGMuxer(StreamIO):
 
         self._cmd = [self.command(session), '-nostats', '-y']
         for np in self.pipes:
-            self._cmd.extend(["-i", np.path])
+            self._cmd.extend(["-i", str(np.path)])
 
         self._cmd.extend(['-c:v', videocodec])
         self._cmd.extend(['-c:a', audiocodec])
diff --git a/src/streamlink/utils/named_pipe.py b/src/streamlink/utils/named_pipe.py
index f4b1b874..b102d3b5 100644
--- a/src/streamlink/utils/named_pipe.py
+++ b/src/streamlink/utils/named_pipe.py
@@ -1,69 +1,124 @@
+import abc
+import logging
 import os
+import random
 import tempfile
+import threading
+from pathlib import Path
 
 from streamlink.compat import is_win32
 
 if is_win32:
     from ctypes import windll, cast, c_ulong, c_void_p, byref
 
-    PIPE_ACCESS_OUTBOUND = 0x00000002
-    PIPE_TYPE_BYTE = 0x00000000
-    PIPE_READMODE_BYTE = 0x00000000
-    PIPE_WAIT = 0x00000000
-    PIPE_UNLIMITED_INSTANCES = 255
-    INVALID_HANDLE_VALUE = -1
+
+log = logging.getLogger(__name__)
+
+_lock = threading.Lock()
+_id = 0
 
 
-class NamedPipe:
-    def __init__(self, name):
-        self.fifo = None
-        self.pipe = None
+class NamedPipeBase(abc.ABC):
+    path: Path
 
-        if is_win32:
-            self.path = os.path.join("\\\\.\\pipe", name)
-            self.pipe = self._create_named_pipe(self.path)
-        else:
-            self.path = os.path.join(tempfile.gettempdir(), name)
-            self._create_fifo(self.path)
+    def __init__(self):
+        global _id
+        with _lock:
+            _id += 1
+            self.name = f"streamlinkpipe-{os.getpid()}-{_id}-{random.randint(0, 9999)}"
+        log.info(f"Creating pipe {self.name}")
+        self._create()
 
-    def _create_fifo(self, name):
-        os.mkfifo(name, 0o660)
+    @abc.abstractmethod
+    def _create(self) -> None:
+        raise NotImplementedError
 
-    def _create_named_pipe(self, path):
-        bufsize = 8192
+    @abc.abstractmethod
+    def open(self) -> None:
+        raise NotImplementedError
 
-        create_named_pipe = windll.kernel32.CreateNamedPipeW
+    @abc.abstractmethod
+    def write(self, data) -> int:
+        raise NotImplementedError
 
-        pipe = create_named_pipe(path, PIPE_ACCESS_OUTBOUND,
-                                 PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
-                                 PIPE_UNLIMITED_INSTANCES,
-                                 bufsize, bufsize,
-                                 0, None)
+    @abc.abstractmethod
+    def close(self) -> None:
+        raise NotImplementedError
 
-        if pipe == INVALID_HANDLE_VALUE:
-            error_code = windll.kernel32.GetLastError()
-            raise OSError("Error code 0x{0:08X}".format(error_code))
 
-        return pipe
+class NamedPipePosix(NamedPipeBase):
+    mode = "wb"
+    permissions = 0o660
+    fifo = None
 
-    def open(self, mode):
-        if not self.pipe:
-            self.fifo = open(self.path, mode)
+    def _create(self):
+        self.path = Path(tempfile.gettempdir(), self.name)
+        os.mkfifo(self.path, self.permissions)
+
+    def open(self):
+        self.fifo = open(self.path, self.mode)
 
     def write(self, data):
-        if self.pipe:
-            windll.kernel32.ConnectNamedPipe(self.pipe, None)
-            written = c_ulong(0)
-            windll.kernel32.WriteFile(self.pipe, cast(data, c_void_p),
-                                      len(data), byref(written),
-                                      None)
-            return written
-        else:
-            return self.fifo.write(data)
+        return self.fifo.write(data)
 
     def close(self):
-        if self.pipe:
-            windll.kernel32.DisconnectNamedPipe(self.pipe)
-        else:
+        if self.fifo:
             self.fifo.close()
+            self.fifo = None
+        if self.path.is_fifo():
             os.unlink(self.path)
+
+
+class NamedPipeWindows(NamedPipeBase):
+    bufsize = 8192
+    pipe = None
+
+    PIPE_ACCESS_OUTBOUND = 0x00000002
+    PIPE_TYPE_BYTE = 0x00000000
+    PIPE_READMODE_BYTE = 0x00000000
+    PIPE_WAIT = 0x00000000
+    PIPE_UNLIMITED_INSTANCES = 255
+    INVALID_HANDLE_VALUE = -1
+
+    @staticmethod
+    def _get_last_error():
+        error_code = windll.kernel32.GetLastError()
+        raise OSError(f"Named pipe error code 0x{error_code:08X}")
+
+    def _create(self):
+        self.path = Path("\\\\.\\pipe", self.name)
+        self.pipe = windll.kernel32.CreateNamedPipeW(
+            str(self.path),
+            self.PIPE_ACCESS_OUTBOUND,
+            self.PIPE_TYPE_BYTE | self.PIPE_READMODE_BYTE | self.PIPE_WAIT,
+            self.PIPE_UNLIMITED_INSTANCES,
+            self.bufsize,
+            self.bufsize,
+            0,
+            None
+        )
+        if self.pipe == self.INVALID_HANDLE_VALUE:
+            self._get_last_error()
+
+    def open(self):
+        windll.kernel32.ConnectNamedPipe(self.pipe, None)
+
+    def write(self, data):
+        written = c_ulong(0)
+        windll.kernel32.WriteFile(
+            self.pipe,
+            cast(data, c_void_p),
+            len(data),
+            byref(written),
+            None
+        )
+        return written.value
+
+    def close(self):
+        if self.pipe is not None:
+            windll.kernel32.DisconnectNamedPipe(self.pipe)
+            windll.kernel32.CloseHandle(self.pipe)
+            self.pipe = None
+
+
+NamedPipe = NamedPipePosix if not is_win32 else NamedPipeWindows
diff --git a/src/streamlink_cli/main.py b/src/streamlink_cli/main.py
index c48868d2..2db30ac7 100644
--- a/src/streamlink_cli/main.py
+++ b/src/streamlink_cli/main.py
@@ -96,11 +96,8 @@ def create_output(plugin):
                          "executable with --player.")
 
         if args.player_fifo:
-            pipename = "streamlinkpipe-{0}".format(os.getpid())
-            log.info("Creating pipe {0}".format(pipename))
-
             try:
-                namedpipe = NamedPipe(pipename)
+                namedpipe = NamedPipe()
             except OSError as err:
                 console.exit("Failed to create pipe: {0}", err)
         elif args.player_http:
diff --git a/src/streamlink_cli/output.py b/src/streamlink_cli/output.py
index f684625e..24169014 100644
--- a/src/streamlink_cli/output.py
+++ b/src/streamlink_cli/output.py
@@ -270,7 +270,7 @@ class PlayerOutput(Output):
             raise OSError("Process exited prematurely")
 
         if self.namedpipe:
-            self.namedpipe.open("wb")
+            self.namedpipe.open()
         elif self.http:
             self.http.open()
 


Test patch:
diff --git a/tests/test_utils_named_pipe.py b/tests/test_utils_named_pipe.py
new file mode 100644
index 00000000..7b78bd38
--- /dev/null
+++ b/tests/test_utils_named_pipe.py
@@ -0,0 +1,164 @@
+import threading
+import unittest
+from unittest.mock import Mock, call, patch
+
+from streamlink.compat import is_win32
+from streamlink.utils.named_pipe import NamedPipe, NamedPipePosix, NamedPipeWindows
+
+if is_win32:
+    from ctypes import windll, create_string_buffer, c_ulong, byref
+
+
+GENERIC_READ = 0x80000000
+OPEN_EXISTING = 3
+
+
+class ReadNamedPipeThread(threading.Thread):
+    def __init__(self, pipe: NamedPipe):
+        super().__init__(daemon=True)
+        self.path = str(pipe.path)
+        self.error = None
+        self.data = b""
+        self.done = threading.Event()
+
+    def run(self):
+        try:
+            self.read()
+        except OSError as err:  # pragma: no cover
+            self.error = err
+        self.done.set()
+
+    def read(self):
+        raise NotImplementedError
+
+
+class ReadNamedPipeThreadPosix(ReadNamedPipeThread):
+    def read(self):
+        with open(self.path, "rb") as file:
+            while True:
+                data = file.read(-1)
+                if len(data) == 0:
+                    break
+                self.data += data
+
+
+class ReadNamedPipeThreadWindows(ReadNamedPipeThread):
+    def read(self):
+        handle = windll.kernel32.CreateFileW(self.path, GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        try:
+            while True:
+                data = create_string_buffer(NamedPipeWindows.bufsize)
+                read = c_ulong(0)
+                if not windll.kernel32.ReadFile(handle, data, NamedPipeWindows.bufsize, byref(read), None):  # pragma: no cover
+                    raise OSError(f"Failed reading pipe: {windll.kernel32.GetLastError()}")
+                self.data += data.value
+                if read.value != len(data.value):
+                    break
+        finally:
+            windll.kernel32.CloseHandle(handle)
+
+
+class TestNamedPipe(unittest.TestCase):
+    @patch("streamlink.utils.named_pipe._id", 0)
+    @patch("streamlink.utils.named_pipe.os.getpid", Mock(return_value=12345))
+    @patch("streamlink.utils.named_pipe.random.randint", Mock(return_value=67890))
+    @patch("streamlink.utils.named_pipe.NamedPipe._create", Mock(return_value=None))
+    @patch("streamlink.utils.named_pipe.log")
+    def test_name(self, mock_log):
+        NamedPipe()
+        NamedPipe()
+        self.assertEqual(mock_log.info.mock_calls, [
+            call("Creating pipe streamlinkpipe-12345-1-67890"),
+            call("Creating pipe streamlinkpipe-12345-2-67890")
+        ])
+
+
+@unittest.skipIf(is_win32, "test only applicable on a POSIX OS")
+class TestNamedPipePosix(unittest.TestCase):
+    def test_export(self):
+        self.assertEqual(NamedPipe, NamedPipePosix)
+
+    @patch("streamlink.utils.named_pipe.os.mkfifo")
+    def test_create(self, mock_mkfifo):
+        mock_mkfifo.side_effect = OSError()
+        with self.assertRaises(OSError):
+            NamedPipePosix()
+        self.assertEqual(mock_mkfifo.call_args[0][1:], (0o660,))
+
+    def test_close_before_open(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        pipe.close()
+        self.assertFalse(pipe.path.is_fifo())
+        # closing twice doesn't raise
+        pipe.close()
+
+    def test_write_before_open(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        with self.assertRaises(Exception):
+            pipe.write(b"foo")
+        pipe.close()
+
+    def test_named_pipe(self):
+        pipe = NamedPipePosix()
+        self.assertTrue(pipe.path.is_fifo())
+        reader = ReadNamedPipeThreadPosix(pipe)
+        reader.start()
+        pipe.open()
+        self.assertEqual(pipe.write(b"foo"), 3)
+        self.assertEqual(pipe.write(b"bar"), 3)
+        pipe.close()
+        self.assertFalse(pipe.path.is_fifo())
+        reader.done.wait(4000)
+        self.assertEqual(reader.error, None)
+        self.assertEqual(reader.data, b"foobar")
+        self.assertFalse(reader.is_alive())
+
+
+@unittest.skipIf(not is_win32, "test only applicable on Windows")
+class TestNamedPipeWindows(unittest.TestCase):
+    def test_export(self):
+        self.assertEqual(NamedPipe, NamedPipeWindows)
+
+    @patch("streamlink.utils.named_pipe.windll.kernel32")
+    def test_create(self, mock_kernel32):
+        mock_kernel32.CreateNamedPipeW.return_value = NamedPipeWindows.INVALID_HANDLE_VALUE
+        mock_kernel32.GetLastError.return_value = 12345
+        with self.assertRaises(OSError) as cm:
+            NamedPipeWindows()
+        self.assertEqual(str(cm.exception), "Named pipe error code 0x00003039")
+        self.assertEqual(mock_kernel32.CreateNamedPipeW.call_args[0][1:], (
+            0x00000002,
+            0x00000000,
+            255,
+            8192,
+            8192,
+            0,
+            None
+        ))
+
+    def test_close_before_open(self):
+        pipe = NamedPipeWindows()
+        handle = windll.kernel32.CreateFileW(str(pipe.path), GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        self.assertNotEqual(handle, NamedPipeWindows.INVALID_HANDLE_VALUE)
+        windll.kernel32.CloseHandle(handle)
+        pipe.close()
+        handle = windll.kernel32.CreateFileW(str(pipe.path), GENERIC_READ, 0, None, OPEN_EXISTING, 0, None)
+        self.assertEqual(handle, NamedPipeWindows.INVALID_HANDLE_VALUE)
+        # closing twice doesn't raise
+        pipe.close()
+
+    def test_named_pipe(self):
+        pipe = NamedPipeWindows()
+        reader = ReadNamedPipeThreadWindows(pipe)
+        reader.start()
+        pipe.open()
+        self.assertEqual(pipe.write(b"foo"), 3)
+        self.assertEqual(pipe.write(b"bar"), 3)
+        self.assertEqual(pipe.write(b"\0"), 1)
+        reader.done.wait(4000)
+        self.assertEqual(reader.error, None)
+        self.assertEqual(reader.data, b"foobar")
+        self.assertFalse(reader.is_alive())
+        pipe.close()