dom/websocket/tests/file_websocket_permessage_deflate_wsh.py

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