prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Send keepalive messages on websocket to prevent cloudflare disconnect
  buildbot behind cloudflare is currently reloading for me every 100 seconds. This is due to a cloudflare limitation that is resetting websockets with 100 seconds of inactivity. It seems the only ways around that is either an upgrade to their enterprise plan or adding keepalive messages to the application.
  
  https://community.cloudflare.com/t/cloudflare-websocket-timeout/5865
  
  I'm on buildbot 1.0.0 due to openbsd -stable, I've grepped through the source on master and it seems there's already a ping command implemented on the server, but it's not automatically used by the client yet. The client should invoke it every 60 seconds to avoid a disconnect due to timeouts.
  
  Thanks!
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `master/buildbot/www/ws.py`
  Content:
  ```
  1 # This file is part of .  Buildbot is free software: you can
  2 # redistribute it and/or modify it under the terms of the GNU General Public
  3 # License as published by the Free Software Foundation, version 2.
  4 #
  5 # This program is distributed in the hope that it will be useful, but WITHOUT
  6 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  7 # FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  8 # details.
  9 #
  10 # You should have received a copy of the GNU General Public License along with
  11 # this program; if not, write to the Free Software Foundation, Inc., 51
  12 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  13 #
  14 # Copyright  Team Members
  15 
  16 import json
  17 
  18 from autobahn.twisted.resource import WebSocketResource
  19 from autobahn.twisted.websocket import WebSocketServerFactory
  20 from autobahn.twisted.websocket import WebSocketServerProtocol
  21 from twisted.internet import defer
  22 from twisted.python import log
  23 
  24 from buildbot.util import bytes2unicode
  25 from buildbot.util import toJson
  26 from buildbot.util import unicode2bytes
  27 
  28 
  29 class WsProtocol(WebSocketServerProtocol):
  30 
  31     def __init__(self, master):
  32         super().__init__()
  33         self.master = master
  34         self.qrefs = {}
  35         self.debug = self.master.config.www.get('debug', False)
  36 
  37     def sendJsonMessage(self, **msg):
  38         return self.sendMessage(unicode2bytes(json.dumps(msg, default=toJson,
  39                                                          separators=(',', ':'))))
  40 
  41     def onMessage(self, frame, isBinary):
  42         if self.debug:
  43             log.msg("FRAME {}".format(frame))
  44         # parse the incoming request
  45 
  46         frame = json.loads(bytes2unicode(frame))
  47         _id = frame.get("_id")
  48         if _id is None:
  49             return self.sendJsonMessage(error="no '_id' in websocket frame", code=400, _id=None)
  50         cmd = frame.pop("cmd", None)
  51         if cmd is None:
  52             return self.sendJsonMessage(error="no 'cmd' in websocket frame", code=400, _id=None)
  53         cmdmeth = "cmd_" + cmd
  54         meth = getattr(self, cmdmeth, None)
  55         if meth is None:
  56             return self.sendJsonMessage(error="no such command '{}'".format(cmd), code=404, _id=_id)
  57         try:
  58             return meth(**frame)
  59         except TypeError as e:
  60             return self.sendJsonMessage(error="Invalid method argument '{}'".format(str(e)),
  61                                         code=400, _id=_id)
  62         except Exception as e:
  63             log.err("while calling command {}".format(cmd))
  64             return self.sendJsonMessage(error="Internal Error '{}'".format(str(e)), code=500,
  65                                         _id=_id)
  66 
  67     def ack(self, _id):
  68         return self.sendJsonMessage(msg="OK", code=200, _id=_id)
  69 
  70     def parsePath(self, path):
  71         path = path.split("/")
  72         return tuple([str(p) if p != "*" else None for p in path])
  73 
  74     def isPath(self, path):
  75         if not isinstance(path, str):
  76             return False
  77         return True
  78 
  79     @defer.inlineCallbacks
  80     def cmd_startConsuming(self, path, _id):
  81         if not self.isPath(path):
  82             yield self.sendJsonMessage(error="invalid path format '{}'".format(str(path)), code=400,
  83                                        _id=_id)
  84             return
  85 
  86         # if it's already subscribed, don't leak a subscription
  87         if self.qrefs is not None and path in self.qrefs:
  88             yield self.ack(_id=_id)
  89             return
  90 
  91         def callback(key, message):
  92             # protocol is deliberately concise in size
  93             return self.sendJsonMessage(k="/".join(key), m=message)
  94 
  95         qref = yield self.master.mq.startConsuming(callback, self.parsePath(path))
  96 
  97         # race conditions handling
  98         if self.qrefs is None or path in self.qrefs:
  99             qref.stopConsuming()
  100 
  101         # only store and ack if we were not disconnected in between
  102         if self.qrefs is not None:
  103             self.qrefs[path] = qref
  104             self.ack(_id=_id)
  105 
  106     @defer.inlineCallbacks
  107     def cmd_stopConsuming(self, path, _id):
  108         if not self.isPath(path):
  109             yield self.sendJsonMessage(error="invalid path format '{}'".format(str(path)), code=400,
  110                                        _id=_id)
  111             return
  112 
  113         # only succeed if path has been started
  114         if path in self.qrefs:
  115             qref = self.qrefs.pop(path)
  116             yield qref.stopConsuming()
  117             yield self.ack(_id=_id)
  118             return
  119         yield self.sendJsonMessage(error="path was not consumed '{}'".format(str(path)), code=400,
  120                                    _id=_id)
  121 
  122     def cmd_ping(self, _id):
  123         self.sendJsonMessage(msg="pong", code=200, _id=_id)
  124 
  125     def connectionLost(self, reason):
  126         if self.debug:
  127             log.msg("connection lost", system=self)
  128         for qref in self.qrefs.values():
  129             qref.stopConsuming()
  130         self.qrefs = None  # to be sure we don't add any more
  131 
  132 
  133 class WsProtocolFactory(WebSocketServerFactory):
  134 
  135     def __init__(self, master):
  136         super().__init__()
  137         self.master = master
  138         self.setProtocolOptions(webStatus=False)
  139 
  140     def buildProtocol(self, addr):
  141         p = WsProtocol(self.master)
  142         p.factory = self
  143         return p
  144 
  145 
  146 class WsResource(WebSocketResource):
  147 
  148     def __init__(self, master):
  149         super().__init__(WsProtocolFactory(master))
  150
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/master/buildbot/www/ws.py b/master/buildbot/www/ws.py
  --- a/master/buildbot/www/ws.py
  +++ b/master/buildbot/www/ws.py
  @@ -135,7 +135,8 @@
       def __init__(self, master):
           super().__init__()
           self.master = master
  -        self.setProtocolOptions(webStatus=False)
  +        pingInterval = self.master.config.www.get('ws_ping_interval', 0)
  +        self.setProtocolOptions(webStatus=False, autoPingInterval=pingInterval)
   
       def buildProtocol(self, addr):
           p = WsProtocol(self.master)
  
verification_info: {"golden_diff": "diff --git a/master/buildbot/www/ws.py b/master/buildbot/www/ws.py\n--- a/master/buildbot/www/ws.py\n+++ b/master/buildbot/www/ws.py\n@@ -135,7 +135,8 @@\n     def __init__(self, master):\n         super().__init__()\n         self.master = master\n-        self.setProtocolOptions(webStatus=False)\n+        pingInterval = self.master.config.www.get('ws_ping_interval', 0)\n+        self.setProtocolOptions(webStatus=False, autoPingInterval=pingInterval)\n \n     def buildProtocol(self, addr):\n         p = WsProtocol(self.master)\n", "issue": "Send keepalive messages on websocket to prevent cloudflare disconnect\nbuildbot behind cloudflare is currently reloading for me every 100 seconds. This is due to a cloudflare limitation that is resetting websockets with 100 seconds of inactivity. It seems the only ways around that is either an upgrade to their enterprise plan or adding keepalive messages to the application.\r\n\r\nhttps://community.cloudflare.com/t/cloudflare-websocket-timeout/5865\r\n\r\nI'm on buildbot 1.0.0 due to openbsd -stable, I've grepped through the source on master and it seems there's already a ping command implemented on the server, but it's not automatically used by the client yet. The client should invoke it every 60 seconds to avoid a disconnect due to timeouts.\r\n\r\nThanks!\n", "before_files": [{"content": "# This file is part of .  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright  Team Members\n\nimport json\n\nfrom autobahn.twisted.resource import WebSocketResource\nfrom autobahn.twisted.websocket import WebSocketServerFactory\nfrom autobahn.twisted.websocket import WebSocketServerProtocol\nfrom twisted.internet import defer\nfrom twisted.python import log\n\nfrom buildbot.util import bytes2unicode\nfrom buildbot.util import toJson\nfrom buildbot.util import unicode2bytes\n\n\nclass WsProtocol(WebSocketServerProtocol):\n\n    def __init__(self, master):\n        super().__init__()\n        self.master = master\n        self.qrefs = {}\n        self.debug = self.master.config.www.get('debug', False)\n\n    def sendJsonMessage(self, **msg):\n        return self.sendMessage(unicode2bytes(json.dumps(msg, default=toJson,\n                                                         separators=(',', ':'))))\n\n    def onMessage(self, frame, isBinary):\n        if self.debug:\n            log.msg(\"FRAME {}\".format(frame))\n        # parse the incoming request\n\n        frame = json.loads(bytes2unicode(frame))\n        _id = frame.get(\"_id\")\n        if _id is None:\n            return self.sendJsonMessage(error=\"no '_id' in websocket frame\", code=400, _id=None)\n        cmd = frame.pop(\"cmd\", None)\n        if cmd is None:\n            return self.sendJsonMessage(error=\"no 'cmd' in websocket frame\", code=400, _id=None)\n        cmdmeth = \"cmd_\" + cmd\n        meth = getattr(self, cmdmeth, None)\n        if meth is None:\n            return self.sendJsonMessage(error=\"no such command '{}'\".format(cmd), code=404, _id=_id)\n        try:\n            return meth(**frame)\n        except TypeError as e:\n            return self.sendJsonMessage(error=\"Invalid method argument '{}'\".format(str(e)),\n                                        code=400, _id=_id)\n        except Exception as e:\n            log.err(\"while calling command {}\".format(cmd))\n            return self.sendJsonMessage(error=\"Internal Error '{}'\".format(str(e)), code=500,\n                                        _id=_id)\n\n    def ack(self, _id):\n        return self.sendJsonMessage(msg=\"OK\", code=200, _id=_id)\n\n    def parsePath(self, path):\n        path = path.split(\"/\")\n        return tuple([str(p) if p != \"*\" else None for p in path])\n\n    def isPath(self, path):\n        if not isinstance(path, str):\n            return False\n        return True\n\n    @defer.inlineCallbacks\n    def cmd_startConsuming(self, path, _id):\n        if not self.isPath(path):\n            yield self.sendJsonMessage(error=\"invalid path format '{}'\".format(str(path)), code=400,\n                                       _id=_id)\n            return\n\n        # if it's already subscribed, don't leak a subscription\n        if self.qrefs is not None and path in self.qrefs:\n            yield self.ack(_id=_id)\n            return\n\n        def callback(key, message):\n            # protocol is deliberately concise in size\n            return self.sendJsonMessage(k=\"/\".join(key), m=message)\n\n        qref = yield self.master.mq.startConsuming(callback, self.parsePath(path))\n\n        # race conditions handling\n        if self.qrefs is None or path in self.qrefs:\n            qref.stopConsuming()\n\n        # only store and ack if we were not disconnected in between\n        if self.qrefs is not None:\n            self.qrefs[path] = qref\n            self.ack(_id=_id)\n\n    @defer.inlineCallbacks\n    def cmd_stopConsuming(self, path, _id):\n        if not self.isPath(path):\n            yield self.sendJsonMessage(error=\"invalid path format '{}'\".format(str(path)), code=400,\n                                       _id=_id)\n            return\n\n        # only succeed if path has been started\n        if path in self.qrefs:\n            qref = self.qrefs.pop(path)\n            yield qref.stopConsuming()\n            yield self.ack(_id=_id)\n            return\n        yield self.sendJsonMessage(error=\"path was not consumed '{}'\".format(str(path)), code=400,\n                                   _id=_id)\n\n    def cmd_ping(self, _id):\n        self.sendJsonMessage(msg=\"pong\", code=200, _id=_id)\n\n    def connectionLost(self, reason):\n        if self.debug:\n            log.msg(\"connection lost\", system=self)\n        for qref in self.qrefs.values():\n            qref.stopConsuming()\n        self.qrefs = None  # to be sure we don't add any more\n\n\nclass WsProtocolFactory(WebSocketServerFactory):\n\n    def __init__(self, master):\n        super().__init__()\n        self.master = master\n        self.setProtocolOptions(webStatus=False)\n\n    def buildProtocol(self, addr):\n        p = WsProtocol(self.master)\n        p.factory = self\n        return p\n\n\nclass WsResource(WebSocketResource):\n\n    def __init__(self, master):\n        super().__init__(WsProtocolFactory(master))\n", "path": "master/buildbot/www/ws.py"}], "after_files": [{"content": "# This file is part of .  Buildbot is free software: you can\n# redistribute it and/or modify it under the terms of the GNU General Public\n# License as published by the Free Software Foundation, version 2.\n#\n# This program is distributed in the hope that it will be useful, but WITHOUT\n# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS\n# FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more\n# details.\n#\n# You should have received a copy of the GNU General Public License along with\n# this program; if not, write to the Free Software Foundation, Inc., 51\n# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.\n#\n# Copyright  Team Members\n\nimport json\n\nfrom autobahn.twisted.resource import WebSocketResource\nfrom autobahn.twisted.websocket import WebSocketServerFactory\nfrom autobahn.twisted.websocket import WebSocketServerProtocol\nfrom twisted.internet import defer\nfrom twisted.python import log\n\nfrom buildbot.util import bytes2unicode\nfrom buildbot.util import toJson\nfrom buildbot.util import unicode2bytes\n\n\nclass WsProtocol(WebSocketServerProtocol):\n\n    def __init__(self, master):\n        super().__init__()\n        self.master = master\n        self.qrefs = {}\n        self.debug = self.master.config.www.get('debug', False)\n\n    def sendJsonMessage(self, **msg):\n        return self.sendMessage(unicode2bytes(json.dumps(msg, default=toJson,\n                                                         separators=(',', ':'))))\n\n    def onMessage(self, frame, isBinary):\n        if self.debug:\n            log.msg(\"FRAME {}\".format(frame))\n        # parse the incoming request\n\n        frame = json.loads(bytes2unicode(frame))\n        _id = frame.get(\"_id\")\n        if _id is None:\n            return self.sendJsonMessage(error=\"no '_id' in websocket frame\", code=400, _id=None)\n        cmd = frame.pop(\"cmd\", None)\n        if cmd is None:\n            return self.sendJsonMessage(error=\"no 'cmd' in websocket frame\", code=400, _id=None)\n        cmdmeth = \"cmd_\" + cmd\n        meth = getattr(self, cmdmeth, None)\n        if meth is None:\n            return self.sendJsonMessage(error=\"no such command '{}'\".format(cmd), code=404, _id=_id)\n        try:\n            return meth(**frame)\n        except TypeError as e:\n            return self.sendJsonMessage(error=\"Invalid method argument '{}'\".format(str(e)),\n                                        code=400, _id=_id)\n        except Exception as e:\n            log.err(\"while calling command {}\".format(cmd))\n            return self.sendJsonMessage(error=\"Internal Error '{}'\".format(str(e)), code=500,\n                                        _id=_id)\n\n    def ack(self, _id):\n        return self.sendJsonMessage(msg=\"OK\", code=200, _id=_id)\n\n    def parsePath(self, path):\n        path = path.split(\"/\")\n        return tuple([str(p) if p != \"*\" else None for p in path])\n\n    def isPath(self, path):\n        if not isinstance(path, str):\n            return False\n        return True\n\n    @defer.inlineCallbacks\n    def cmd_startConsuming(self, path, _id):\n        if not self.isPath(path):\n            yield self.sendJsonMessage(error=\"invalid path format '{}'\".format(str(path)), code=400,\n                                       _id=_id)\n            return\n\n        # if it's already subscribed, don't leak a subscription\n        if self.qrefs is not None and path in self.qrefs:\n            yield self.ack(_id=_id)\n            return\n\n        def callback(key, message):\n            # protocol is deliberately concise in size\n            return self.sendJsonMessage(k=\"/\".join(key), m=message)\n\n        qref = yield self.master.mq.startConsuming(callback, self.parsePath(path))\n\n        # race conditions handling\n        if self.qrefs is None or path in self.qrefs:\n            qref.stopConsuming()\n\n        # only store and ack if we were not disconnected in between\n        if self.qrefs is not None:\n            self.qrefs[path] = qref\n            self.ack(_id=_id)\n\n    @defer.inlineCallbacks\n    def cmd_stopConsuming(self, path, _id):\n        if not self.isPath(path):\n            yield self.sendJsonMessage(error=\"invalid path format '{}'\".format(str(path)), code=400,\n                                       _id=_id)\n            return\n\n        # only succeed if path has been started\n        if path in self.qrefs:\n            qref = self.qrefs.pop(path)\n            yield qref.stopConsuming()\n            yield self.ack(_id=_id)\n            return\n        yield self.sendJsonMessage(error=\"path was not consumed '{}'\".format(str(path)), code=400,\n                                   _id=_id)\n\n    def cmd_ping(self, _id):\n        self.sendJsonMessage(msg=\"pong\", code=200, _id=_id)\n\n    def connectionLost(self, reason):\n        if self.debug:\n            log.msg(\"connection lost\", system=self)\n        for qref in self.qrefs.values():\n            qref.stopConsuming()\n        self.qrefs = None  # to be sure we don't add any more\n\n\nclass WsProtocolFactory(WebSocketServerFactory):\n\n    def __init__(self, master):\n        super().__init__()\n        self.master = master\n        pingInterval = self.master.config.www.get('ws_ping_interval', 0)\n        self.setProtocolOptions(webStatus=False, autoPingInterval=pingInterval)\n\n    def buildProtocol(self, addr):\n        p = WsProtocol(self.master)\n        p.factory = self\n        return p\n\n\nclass WsResource(WebSocketResource):\n\n    def __init__(self, master):\n        super().__init__(WsProtocolFactory(master))\n", "path": "master/buildbot/www/ws.py"}]}