third_party/WebKit/LayoutTests/http/tests/websocket/permessage-deflate-split-frames_wsh.py

   1 # Copyright 2014 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5
   6 from mod_pywebsocket import common
   7 from mod_pywebsocket.extensions import PerMessageDeflateExtensionProcessor
   8 from mod_pywebsocket.stream import create_header
   9
  10
  11 def _get_permessage_deflate_extension_processor(request):
  12     for extension_processor in request.ws_extension_processors:
  13         if isinstance(extension_processor,
  14                       PerMessageDeflateExtensionProcessor):
  15             return extension_processor
  16     return None
  17
  18
  19 def web_socket_do_extra_handshake(request):
  20     processor = _get_permessage_deflate_extension_processor(request)
  21     assert processor is not None
  22     # Remove extension processors other than
  23     # PerMessageDeflateExtensionProcessor to avoid conflict.
  24     request.ws_extension_processors = [processor]
  25
  26
  27 def web_socket_transfer_data(request):
  28     line = request.ws_stream.receive_message()
  29     # Hello
  30     payload = b'\xf2\x48\xcd\xc9\xc9\x07\x00\x00\x00\xff\xff'
  31     # Strip \x00\x00\xff\xff
  32     stripped = payload[:-4]
  33
  34     header = create_header(common.OPCODE_TEXT, len(payload),
  35         fin=0, rsv1=1, rsv2=0, rsv3=0, mask=False)
  36     request.ws_stream._write(header + payload)
  37
  38     header = create_header(common.OPCODE_CONTINUATION, len(stripped),
  39         fin=1, rsv1=0, rsv2=0, rsv3=0, mask=False)
  40     request.ws_stream._write(header + stripped)
  41
  42
  43 # vi:sts=4 sw=4 et