LogDog: Properly emit Butler stream magic.

client/libs/logdog/stream.py

 # Copyright 2016 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 import collections
 import contextlib
 import json
 import os
 import socket
 import sys
 import threading
 import types
 
 from libs.logdog import streamname, varint
 
 
 _StreamParamsBase = collections.namedtuple('_StreamParamsBase',
     ('name', 'type', 'content_type', 'tags', 'tee', 'binary_file_extension'))
 
 
+# Magic number at the beginning of a Butler stream
+#
+# See "ProtocolFrameHeaderMagic" in:
+# <luci-go>/logdog/client/butlerlib/streamproto
+BUTLER_MAGIC = 'BTLR1\x1e'
+
+
 class StreamParams(_StreamParamsBase):
   """Defines the set of parameters to apply to a new stream."""
 
   # A text content stream.
   TEXT = 'text'
   # A binary content stream.
   BINARY = 'binary'
   # A datagram content stream.
   DATAGRAM = 'datagram'
 
@@ skipping 184 matching lines @@
       params (StreamParams): The parameters to use with the new connection.
 
     Raises:
       ValueError if the stream name has already been used, or if the parameters
       are not valid.
     """
     self._register_new_stream(params.name)
     params_json = params.to_json()
 
     fd = self._connect_raw()
+    fd.write(BUTLER_MAGIC)
     varint.write_uvarint(fd, len(params_json))
     fd.write(params_json)
     return fd
 
   @contextlib.contextmanager
   def text(self, name, **kwargs):
     """Context manager to create, use, and teardown a TEXT stream.
 
     This context manager creates a new butler TEXT stream with the specified
     parameters, yields it, and closes it on teardown.
@@ skipping 169 matching lines @@
   def _connect_raw(self):
     return open(self._name, 'wb')
 
 _default_registry.register_protocol('net.pipe', _NamedPipeStreamClient)
 
 
 class _UnixDomainSocketStreamClient(StreamClient):
   """A StreamClient implementation that uses a UNIX domain socket.
   """
 
+  class SocketFile(object):
+    """A write-only file-like object that writes to a UNIX socket."""
+
+    def __init__(self, fd):
+      self._fd = fd
+
+    def write(self, data):
+      self._fd.send(data)
+
+    def close(self):
+      self._fd.close()
+
+
   def __init__(self, path):
     """Initializes a new UNIX domain socket stream client.
 
     Args:
       path (str): The path to the named UNIX domain socket.
     """
     super(_UnixDomainSocketStreamClient, self).__init__()
     self._path = path
 
   @classmethod
   def _create(cls, value):
     if not os.path.exists(value):
       raise ValueError('UNIX domain socket [%s] does not exist.' % (value,))
     return cls(value)
 
   def _connect_raw(self):
     sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
     sock.connect(self._path)
-    return sock
+    return self.SocketFile(sock)
 
 _default_registry.register_protocol('unix', _UnixDomainSocketStreamClient)