February 2014 - calendarserver-changes

[12777] CalendarServer/trunk/calendarserver/webadmin
by source_changes＠macosforge.org 28 Feb '14

28 Feb '14

Revision: 12777 http://trac.calendarserver.org//changeset/12777 Author: wsanchez(a)apple.com Date: 2014-02-27 18:29:10 -0800 (Thu, 27 Feb 2014) Log Message: ----------- poll in render() Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/work.py CalendarServer/trunk/calendarserver/webadmin/work.xhtml Modified: CalendarServer/trunk/calendarserver/webadmin/work.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/work.py 2014-02-28 02:02:46 UTC (rev 12776) +++ CalendarServer/trunk/calendarserver/webadmin/work.py 2014-02-28 02:29:10 UTC (rev 12777) @@ -29,7 +29,7 @@ from zope.interface import implementer -from twisted.internet.defer import inlineCallbacks # , returnValue +from twisted.internet.defer import inlineCallbacks, returnValue # from twisted.web.template import tags as html, renderer # from txdav.caldav.datastore.scheduling.imip.inbound import ( @@ -94,6 +94,12 @@ @inlineCallbacks + def render(self, request): + yield self.poll() + returnValue(super(WorkEventsResource, self).render(request)) + + + @inlineCallbacks def poll(self): txn = self._store.newTransaction() @@ -140,21 +146,18 @@ payload[workDescription] = categoryData + self.addEvents(( + dict( + eventClass=u"work", + eventText=asJSON(payload), + ), + )) - print(payload) - - # self._workEventsResource.addEvents(( - # dict( - # eventClass=u"work", - # eventText=asJSON(payload), - # ), - # )) - if not hasattr(self, "_clock"): from twisted.internet import reactor self._clock = reactor - self._clock.callLater(5, self.poll) + # self._clock.callLater(5, self.poll) Modified: CalendarServer/trunk/calendarserver/webadmin/work.xhtml =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/work.xhtml 2014-02-28 02:02:46 UTC (rev 12776) +++ CalendarServer/trunk/calendarserver/webadmin/work.xhtml 2014-02-28 02:29:10 UTC (rev 12777) @@ -83,6 +83,7 @@ window.onload = function() { initChart(); + initEventSource(); }; </script>

1 0

[12776] CalendarServer/trunk/calendarserver/webadmin
by source_changes＠macosforge.org 28 Feb '14

28 Feb '14

Revision: 12776 http://trac.calendarserver.org//changeset/12776 Author: wsanchez(a)apple.com Date: 2014-02-27 18:02:46 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Poll for work activity Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/work.py CalendarServer/trunk/calendarserver/webadmin/work.xhtml Modified: CalendarServer/trunk/calendarserver/webadmin/work.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/work.py 2014-02-28 00:51:41 UTC (rev 12775) +++ CalendarServer/trunk/calendarserver/webadmin/work.py 2014-02-28 02:02:46 UTC (rev 12776) @@ -25,9 +25,26 @@ "WorkMonitorResource", ] -# from twisted.internet.defer import inlineCallbacks, returnValue +from json import dumps + +from zope.interface import implementer + +from twisted.internet.defer import inlineCallbacks # , returnValue # from twisted.web.template import tags as html, renderer +# from txdav.caldav.datastore.scheduling.imip.inbound import ( +# IMIPPollingWork, IMIPReplyWork +# ) + +# from twistedcaldav.directory.directory import GroupCacherPollingWork +# from calendarserver.push.notifier import PushNotificationWork + +from txdav.caldav.datastore.scheduling.work import ( + ScheduleOrganizerWork, ScheduleReplyWork, ScheduleRefreshWork +) + + +from .eventsource import EventSourceResource, IEventDecoder from .resource import PageElement, TemplateResource @@ -37,12 +54,10 @@ Principal management page element. """ - def __init__(self, store): + def __init__(self): PageElement.__init__(self, u"work") - self._store = store - def pageSlots(self): return { u"title": u"Workload Monitor", @@ -60,5 +75,106 @@ def __init__(self, store): TemplateResource.__init__( - self, lambda: WorkMonitorPageElement(store) + self, lambda: WorkMonitorPageElement() ) + + self.putChild(u"events", WorkEventsResource(store)) + + + +class WorkEventsResource(EventSourceResource): + """ + Resource that vends work queue information via HTML5 EventSource events. + """ + + def __init__(self, store): + EventSourceResource.__init__(self, EventDecoder, bufferSize=1) + + self._store = store + + + @inlineCallbacks + def poll(self): + txn = self._store.newTransaction() + + payload = {} + + for workDescription, workItemClass, itemAttributes in ( + ( + u"Organizer Request", + ScheduleOrganizerWork, + ( + ("icalendarUid", "iCalendar UID"), + ("attendeeCount", "Attendee Count"), + ), + ), + ( + u"Attendee Reply", + ScheduleReplyWork, + ( + ("icalendarUid", "iCalendar UID"), + ), + ), + ( + u"Attendee Refresh", + ScheduleRefreshWork, + ( + ("icalendarUid", "iCalendar UID"), + ("attendeeCount", "Attendee Count"), + ), + ), + ): + workItems = yield workItemClass.all(txn) + + categoryData = [] + + for workItem in workItems: + itemData = {} + + for itemAttribute, itemDescription in itemAttributes: + itemData[itemDescription] = getattr( + workItem, itemAttribute + ) + + categoryData.append(itemData) + + payload[workDescription] = categoryData + + + print(payload) + + # self._workEventsResource.addEvents(( + # dict( + # eventClass=u"work", + # eventText=asJSON(payload), + # ), + # )) + + if not hasattr(self, "_clock"): + from twisted.internet import reactor + self._clock = reactor + + self._clock.callLater(5, self.poll) + + + +@implementer(IEventDecoder) +class EventDecoder(object): + @staticmethod + def idForEvent(event): + return event.get("eventID") + + + @staticmethod + def classForEvent(event): + return event.get("eventClass") + + + @staticmethod + def textForEvent(event): + return event.get("eventText") + + + +def asJSON(obj): + return dumps(obj, separators=(',', ':')) Modified: CalendarServer/trunk/calendarserver/webadmin/work.xhtml =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/work.xhtml 2014-02-28 00:51:41 UTC (rev 12775) +++ CalendarServer/trunk/calendarserver/webadmin/work.xhtml 2014-02-28 02:02:46 UTC (rev 12776) @@ -27,9 +27,6 @@ <script> function drawChart(data) { - - // var data = [4, 8, 15, 16, 23, 42]; - var width = 600; var barHeight = 20;

1 0

[12775] CalendarServer/trunk/calendarserver/webadmin/logs.py
by source_changes＠macosforge.org 28 Feb '14

28 Feb '14

Revision: 12775 http://trac.calendarserver.org//changeset/12775 Author: wsanchez(a)apple.com Date: 2014-02-27 16:51:41 -0800 (Thu, 27 Feb 2014) Log Message: ----------- cosmetic Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/logs.py Modified: CalendarServer/trunk/calendarserver/webadmin/logs.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/logs.py 2014-02-28 00:24:11 UTC (rev 12774) +++ CalendarServer/trunk/calendarserver/webadmin/logs.py 2014-02-28 00:51:41 UTC (rev 12775) @@ -68,42 +68,6 @@ -@implementer(IEventDecoder) -class EventDecoder(object): - """ - Decodes logging events. - """ - - @staticmethod - def idForEvent(event): - observer, eventClass, logEvent = event - return id(logEvent) - - - @staticmethod - def classForEvent(event): - observer, eventClass, logEvent = event - return eventClass - - - @staticmethod - def textForEvent(event): - observer, eventClass, logEvent = event - - try: - if eventClass == u"access": - text = logEvent[u"log-format"] % logEvent - else: - text = observer.formatEvent(logEvent) - if text is None: - text = u"" - except: - text = u"*** Error while formatting event ***" - - return text - - - class LogEventsResource(EventSourceResource): """ Log event vending resource. @@ -180,3 +144,39 @@ return self._resource.addEvents(((self, u"access", event),)) + + + +@implementer(IEventDecoder) +class EventDecoder(object): + """ + Decodes logging events. + """ + + @staticmethod + def idForEvent(event): + observer, eventClass, logEvent = event + return id(logEvent) + + + @staticmethod + def classForEvent(event): + observer, eventClass, logEvent = event + return eventClass + + + @staticmethod + def textForEvent(event): + observer, eventClass, logEvent = event + + try: + if eventClass == u"access": + text = logEvent[u"log-format"] % logEvent + else: + text = observer.formatEvent(logEvent) + if text is None: + text = u"" + except: + text = u"*** Error while formatting event ***" + + return text

1 0

[12774] CalendarServer/trunk/calendarserver/webadmin/logs.py
by source_changes＠macosforge.org 28 Feb '14

28 Feb '14

Revision: 12774 http://trac.calendarserver.org//changeset/12774 Author: wsanchez(a)apple.com Date: 2014-02-27 16:24:11 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Use new eventsource module. Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/logs.py Modified: CalendarServer/trunk/calendarserver/webadmin/logs.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/logs.py 2014-02-28 00:23:44 UTC (rev 12773) +++ CalendarServer/trunk/calendarserver/webadmin/logs.py 2014-02-28 00:24:11 UTC (rev 12774) @@ -26,20 +26,13 @@ "LogEventsResource", ] -from collections import deque - from zope.interface import implementer from twisted.python.log import FileLogObserver -from twisted.internet.defer import succeed -from txweb2.stream import IByteStream, fallbackSplit -from txweb2.resource import Resource -from txweb2.http_headers import MimeType -from txweb2.http import Response - from calendarserver.accesslog import CommonAccessLoggingObserverExtensions +from .eventsource import EventSourceResource, IEventDecoder from .resource import PageElement, TemplateResource @@ -75,132 +68,64 @@ -class LogEventsResource(Resource): +@implementer(IEventDecoder) +class EventDecoder(object): """ - Log event vending resource. + Decodes logging events. """ - addSlash = False + @staticmethod + def idForEvent(event): + observer, eventClass, logEvent = event + return id(logEvent) - def __init__(self): - Resource.__init__(self) + @staticmethod + def classForEvent(event): + observer, eventClass, logEvent = event + return eventClass - events = deque(maxlen=400) - observers = ( - AccessLogObserver(events), - ServerLogObserver(events), - ) + @staticmethod + def textForEvent(event): + observer, eventClass, logEvent = event - for observer in observers: - observer.start() + try: + if eventClass == u"access": + text = logEvent[u"log-format"] % logEvent + else: + text = observer.formatEvent(logEvent) + if text is None: + text = u"" + except: + text = u"*** Error while formatting event ***" - self.events = events - self.observers = observers + return text - def render(self, request): - start = request.headers.getRawHeaders(u"last-event-id") - if start is not None: - try: - start = int(start[0]) - except ValueError: - start = None - - response = Response() - response.stream = LogEventStream(self, start=start) - response.headers.setHeader( - b"content-type", MimeType.fromString(b"text/event-stream") - ) - return response - - - -@implementer(IByteStream) -class LogEventStream(object): +class LogEventsResource(EventSourceResource): """ - L{IByteStream} that streams log events out as HTML5 EventSource events. + Log event vending resource. """ - length = None + def __init__(self): + EventSourceResource.__init__(self, EventDecoder) + self.observers = ( + AccessLogObserver(self), + ServerLogObserver(self), + ) - def __init__(self, source, start): - object.__init__(self) + for observer in self.observers: + observer.start() - self._source = source - self._start = start - self._closed = False - def read(self): - if self._closed: - return None - - start = self._start - messageID = None - - for observer, eventClass, event in tuple(self._source.events): - messageID = id(event) - - # If we have a start point, skip messages up to and including the - # one at the start point. - if start is not None: - if messageID == start: - messageID = None - start = None - - continue - - self._start = messageID - - if eventClass == u"access": - message = event[u"log-format"] % event - else: - message = observer.formatEvent(event) - if message is None: - continue - - eventText = textAsEvent( - message, eventID=messageID, eventClass=eventClass - ) - - return succeed(eventText) - - if messageID is not None: - # We just scanned all the messages, and none are the last one the - # client saw. - self._start = None - - marker = "-" - - return succeed( - textAsEvent(marker, eventID=0, eventClass=u"access") + - textAsEvent(marker, eventID=0, eventClass=u"server") - ) - - return succeed(None) - - # from twisted.internet.task import deferLater - # from twisted.internet import reactor - - # return deferLater(reactor, 1.0, self.read) - - - def split(self, point): - return fallbackSplit(self, point) - - - def close(self): - self._closed = True - - - class ServerLogObserver(FileLogObserver): """ - Log observer that captures events in a buffer instead of writing to a file. + Log observer that sends events to an L{EventSourceResource} instead of + writing to a file. @note: L{ServerLogObserver} is an old-style log observer, as it inherits from L{FileLogObserver}. @@ -209,7 +134,7 @@ timeFormat = None - def __init__(self, buffer): + def __init__(self, resource): class FooIO(object): def write(_, s): self._lastMessage = s @@ -220,18 +145,13 @@ FileLogObserver.__init__(self, FooIO()) self.lastMessage = None - self._buffer = buffer - self._waiting = [] + self._resource = resource def emit(self, event): - self._buffer.append((self, u"server", event)) + self._resource.addEvents(((self, u"server", event),)) - while self._waiting: - d = self._waiting.pop(0) - d.callback(None) - def formatEvent(self, event): self._lastMessage = None FileLogObserver.emit(self, event) @@ -248,10 +168,10 @@ ultimately inherits from L{txweb2.log.BaseCommonAccessLoggingObserver}. """ - def __init__(self, buffer): + def __init__(self, resource): CommonAccessLoggingObserverExtensions.__init__(self) - self._buffer = buffer + self._resource = resource def logStats(self, event): @@ -259,21 +179,4 @@ if event[u"type"] != u"access-log": return - self._buffer.append((self, u"access", event)) - - - -def textAsEvent(text, eventID=None, eventClass=None): - event = [] - - if eventID is not None: - event.append(u"id: {0}".format(eventID)) - - if eventClass is not None: - event.append(u"event: {0}".format(eventClass)) - - event.extend( - u"data: {0}".format(l) for l in text.split("\n") - ) - - return u"\n".join(event).encode(u"utf-8") + "\n\n" + self._resource.addEvents(((self, u"access", event),))

1 0

[12773] CalendarServer/trunk/calendarserver/webadmin
by source_changes＠macosforge.org 28 Feb '14

28 Feb '14

Revision: 12773 http://trac.calendarserver.org//changeset/12773 Author: wsanchez(a)apple.com Date: 2014-02-27 16:23:44 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Send EOF when we run out of events; blocking doesn't work when run in server for some reason. Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/eventsource.py CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py Modified: CalendarServer/trunk/calendarserver/webadmin/eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-27 23:46:45 UTC (rev 12772) +++ CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-28 00:23:44 UTC (rev 12773) @@ -153,9 +153,11 @@ ) # Keep track of the event streams - request.addResponseFilter( - lambda r: self._streams.remove(response.stream) - ) + def cleanupFilter(_request, _response): + self._streams.remove(response.stream) + return _response + + request.addResponseFilter(cleanupFilter) self._streams.add(response.stream) return response @@ -240,6 +242,10 @@ return succeed(b"") + # This causes the client to poll, which is undesirable, but the + # deferred below doesn't seem to work in real use... + return succeed(None) + d = Deferred() self._deferredRead = d return d Modified: CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:46:45 UTC (rev 12772) +++ CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-28 00:23:44 UTC (rev 12773) @@ -203,14 +203,25 @@ ) ) - # The next read should block on new events. + + @inlineCallbacks + def test_streamWaitForEvents(self): + """ + Stream reading blocks on additional events. + """ + resource = self.eventSourceResource() + response = self.render(resource) + + # Read should block on new events. d = response.stream.read() self.assertFalse(d.called) d.addErrback(lambda f: None) d.cancel() + test_streamWaitForEvents.todo = "Feature disabled; needs debugging" + @inlineCallbacks def test_streamNewEvents(self): """ @@ -260,6 +271,7 @@ d.addErrback(lambda f: None) d.cancel() + test_streamNewEvents.todo = "Feature disabled; needs debugging" @implementer(IEventDecoder)

1 0

[12772] CalendarServer/trunk/calendarserver/webadmin/eventsource.py
by source_changes＠macosforge.org 27 Feb '14

27 Feb '14

Revision: 12772 http://trac.calendarserver.org//changeset/12772 Author: wsanchez(a)apple.com Date: 2014-02-27 15:46:45 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/eventsource.py Modified: CalendarServer/trunk/calendarserver/webadmin/eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-27 23:28:30 UTC (rev 12771) +++ CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-27 23:46:45 UTC (rev 12772) @@ -250,5 +250,4 @@ def close(self): - print("************ CLOSE ************") self._closed = True

1 0

[12771] CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource. py
by source_changes＠macosforge.org 27 Feb '14

27 Feb '14

Revision: 12771 http://trac.calendarserver.org//changeset/12771 Author: wsanchez(a)apple.com Date: 2014-02-27 15:28:30 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py Modified: CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:23:53 UTC (rev 12770) +++ CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:28:30 UTC (rev 12771) @@ -262,9 +262,6 @@ - # Test closed - - @implementer(IEventDecoder) class DictionaryEventDecoder(object): """

1 0

[12770] CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource. py
by source_changes＠macosforge.org 27 Feb '14

27 Feb '14

Revision: 12770 http://trac.calendarserver.org//changeset/12770 Author: wsanchez(a)apple.com Date: 2014-02-27 15:23:53 -0800 (Thu, 27 Feb 2014) Log Message: ----------- DictionaryEventDecoder is an IEventDecoder Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py Modified: CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:21:22 UTC (rev 12769) +++ CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:23:53 UTC (rev 12770) @@ -20,13 +20,15 @@ from __future__ import print_function +from zope.interface import implementer + from twisted.internet.defer import inlineCallbacks from twisted.trial.unittest import TestCase from txweb2.server import Request from txweb2.http_headers import Headers -from ..eventsource import textAsEvent, EventSourceResource +from ..eventsource import textAsEvent, EventSourceResource, IEventDecoder @@ -263,6 +265,7 @@ # Test closed +@implementer(IEventDecoder) class DictionaryEventDecoder(object): """ Decodes events represented as dictionaries.

1 0

[12769] CalendarServer/trunk/
by source_changes＠macosforge.org 27 Feb '14

27 Feb '14

Revision: 12769 http://trac.calendarserver.org//changeset/12769 Author: wsanchez(a)apple.com Date: 2014-02-27 15:21:22 -0800 (Thu, 27 Feb 2014) Log Message: ----------- ignore subprojects/ Property Changed: ---------------- CalendarServer/trunk/ Property changes on: CalendarServer/trunk ___________________________________________________________________ Modified: svn:ignore - .develop *.egg-info *.pyc data build .settings .project .pydevproject _run + .develop .settings .project .pydevproject *.egg-info *.pyc data build subprojects _run

1 0

[12768] CalendarServer/trunk/calendarserver/webadmin
by source_changes＠macosforge.org 27 Feb '14

27 Feb '14

Revision: 12768 http://trac.calendarserver.org//changeset/12768 Author: wsanchez(a)apple.com Date: 2014-02-27 15:16:58 -0800 (Thu, 27 Feb 2014) Log Message: ----------- Make EventStream.read() defer on new events. Modified Paths: -------------- CalendarServer/trunk/calendarserver/webadmin/eventsource.py CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py Modified: CalendarServer/trunk/calendarserver/webadmin/eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-27 20:17:30 UTC (rev 12767) +++ CalendarServer/trunk/calendarserver/webadmin/eventsource.py 2014-02-27 23:16:58 UTC (rev 12768) @@ -30,7 +30,7 @@ from zope.interface import implementer, Interface -from twisted.internet.defer import succeed +from twisted.internet.defer import Deferred, succeed from txweb2.stream import IByteStream, fallbackSplit from txweb2.resource import Resource @@ -132,12 +132,17 @@ self._eventDecoder = eventDecoder self._events = deque(maxlen=bufferSize) + self._streams = set() - def addEvent(self, event): - self._events.append(event) + def addEvents(self, events): + self._events.extend(events) + # Notify outbound streams that there is new data to vend + for stream in self._streams: + stream.didAddEvents() + def render(self, request): lastID = request.headers.getRawHeaders(u"last-event-id") @@ -146,6 +151,13 @@ response.headers.setHeader( b"content-type", MimeType.fromString(b"text/event-stream") ) + + # Keep track of the event streams + request.addResponseFilter( + lambda r: self._streams.remove(response.stream) + ) + self._streams.add(response.stream) + return response @@ -178,11 +190,19 @@ self._events = events self._lastID = lastID self._closed = False + self._deferredRead = None + def didAddEvents(self): + d = self._deferredRead + if d is not None: + d.addCallback(lambda _: self.read()) + d.callback(None) + + def read(self): if self._closed: - return None + return succeed(None) lastID = self._lastID eventID = None @@ -209,7 +229,7 @@ self._lastID = eventID return succeed( - textAsEvent(eventText, eventID, eventClass).encode("utf-8") + textAsEvent(eventText, eventID, eventClass) ) @@ -218,19 +238,17 @@ # client saw. self._lastID = None - return succeed("") + return succeed(b"") - return succeed(None) + d = Deferred() + self._deferredRead = d + return d - # from twisted.internet.task import deferLater - # from twisted.internet import reactor - # return deferLater(reactor, 1.0, self.read) - - def split(self, point): return fallbackSplit(self, point) def close(self): + print("************ CLOSE ************") self._closed = True Modified: CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py =================================================================== --- CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 20:17:30 UTC (rev 12767) +++ CalendarServer/trunk/calendarserver/webadmin/test/test_eventsource.py 2014-02-27 23:16:58 UTC (rev 12768) @@ -23,7 +23,7 @@ from twisted.internet.defer import inlineCallbacks from twisted.trial.unittest import TestCase -from txweb2.http import Request +from txweb2.server import Request from txweb2.http_headers import Headers from ..eventsource import textAsEvent, EventSourceResource @@ -118,15 +118,10 @@ Tests for L{EventSourceResource}. """ - def eventSourceResource(self, events=()): - resource = EventSourceResource(DictionaryEventDecoder) + def eventSourceResource(self): + return EventSourceResource(DictionaryEventDecoder) - for event in events: - resource.addEvent(event) - return resource - - def render(self, resource): headers = Headers() @@ -190,11 +185,13 @@ dict(eventID=u"4", eventText=u"D"), ) - resource = self.eventSourceResource(events) + resource = self.eventSourceResource() + resource.addEvents(events) + response = self.render(resource) # Each result from read() is another event - for i in range(4): + for i in range(len(events)): result = yield response.stream.read() self.assertEquals( result, @@ -204,11 +201,68 @@ ) ) - result = yield response.stream.read() - self.assertEquals(result, None) + # The next read should block on new events. + d = response.stream.read() + self.assertFalse(d.called) + d.addErrback(lambda f: None) + d.cancel() + @inlineCallbacks + def test_streamNewEvents(self): + """ + Events not already buffered are vended after they are posted. + """ + events = ( + dict(eventID=u"1", eventText=u"A"), + dict(eventID=u"2", eventText=u"B"), + dict(eventID=u"3", eventText=u"C"), + dict(eventID=u"4", eventText=u"D"), + ) + + resource = self.eventSourceResource() + + response = self.render(resource) + + # The first read should block on new events. + d = response.stream.read() + self.assertFalse(d.called) + + # Add some events + resource.addEvents(events) + + # We should now be unblocked + self.assertTrue(d.called) + + # Each result from read() is another event + for i in range(len(events)): + if d is None: + result = yield response.stream.read() + else: + result = yield d + d = None + + self.assertEquals( + result, + textAsEvent( + text=events[i]["eventText"], + eventID=(events[i]["eventID"]) + ) + ) + + # The next read should block on new events. + d = response.stream.read() + self.assertFalse(d.called) + + d.addErrback(lambda f: None) + d.cancel() + + + + # Test closed + + class DictionaryEventDecoder(object): """ Decodes events represented as dictionaries.

1 0