dom/websocket/tests/file_websocket_permessage_deflate_rejected_wsh.py

   1 from mod_pywebsocket import common, msgutil
   2
   3
   4 def web_socket_do_extra_handshake(request):
   5     deflate_removed = False
   6
   7     if request.ws_extension_processors is not None:
   8         for extension_processor in request.ws_extension_processors:
   9             if extension_processor.name() == "deflate":
  10                 request.ws_extension_processors.remove(extension_processor)
  11                 deflate_removed = True
  12
  13     if deflate_removed is False:
  14         raise ValueError("deflate extension processor not found")
  15
  16
  17 def web_socket_transfer_data(request):
  18     while True:
  19         rcvd = msgutil.receive_message(request)
  20         opcode = request.ws_stream.get_last_received_opcode()
  21         if opcode == common.OPCODE_BINARY:
  22             msgutil.send_message(request, rcvd, binary=True)
  23         elif opcode == common.OPCODE_TEXT:
  24             msgutil.send_message(request, rcvd)