Problem
——-
I have two cooperative services in twisted, one is a HTTP proxy and the other a HTTP service to power a web console. At the start and end of every proxy request, an overloaded method copies the request ( headers, POST data, etc ) and also the response. Currently I have a singleton data.Store that receives new records and it works except its becoming unwieldy to unit-test.
Ideas
—–
One thought was to implement an in-process synchronous service bus in a module and make it near stateless.
bus.py
channels = defaultdict(list)
def register(name, callback):
global channels
channels[name].append(callback)
def callbacks(name):
global channels
#use of generator to allow for the caller to apply exception handling logic
for callback in channels[name]:
yield callback
def first(name, default = None):
global channels
return channels[0] if count(channels) > 0 else default
A simple example might be
main.py
store = data.Store()
bus.register("store.addRecord", store.AddRecord)
proxy.py
class ProxyClient(proxy.ProxyClient):
....
def handleResponse(self, data):
proxy.ProxyClient.handleResponse(self, data)
bus.first("store.addRecord", lambda : return False)(data)
Does this seems sensible? bus.channels can be cleared by setUp logic while testing can use in testcase methods for mocking. Or is there already something like this backed into twisted ( which would be much more preferable as the twisted dev’s can usually be relied on to already have unit-testing )
By the time I got to here, I decided that I DID in fact like this solution. Should but may not update to see how this went.