<s> """<STR_LIT>""" <EOL> import json <EOL> import logging <EOL> import threading <EOL> import traceback <EOL> import ufora . FORA . python . ModuleImporter as ModuleImporter <EOL> import ufora . distributed . SharedState . ComputedGraph . SharedStateSynchronizer as SharedStateSynchronizer <EOL> import ufora . distributed . SharedState . ComputedGraph . SynchronousPropertyAccess as SynchronousPropertyAccess <EOL> import ufora . BackendGateway . ComputedGraph . ComputedGraph as ComputedGraph <EOL> import ufora . FORA . VectorDataManager . VectorDataManager as VectorDataManager <EOL> import ufora . BackendGateway . SubscribableWebObjects . AllObjectClassesToExpose as AllObjectClassesToExpose <EOL> import ufora . BackendGateway . SubscribableWebObjects . Exceptions as Exceptions <EOL> import ufora . BackendGateway . SubscribableWebObjects . Decorators as Decorators <EOL> import ufora . BackendGateway . SubscribableWebObjects . Subscriptions as Subscriptions <EOL> import ufora . util . ManagedThread as ManagedThread <EOL> import ufora . util . Unicode as Unicode <EOL> import ufora . FORA . python . Runtime as Runtime <EOL> import ufora . FORA . python . FORA as Fora <EOL> import ufora . FORA . python . ForaValue as ForaValue <EOL> import uuid <EOL> class MalformedMessageException ( Exception ) : <EOL> pass <EOL> class InvalidObjectDefinitionException ( MalformedMessageException ) : <EOL> pass <EOL> class InvalidFieldException ( Exceptions . SubscribableWebObjectsException ) : <EOL> def __init__ ( self ) : <EOL> Exceptions . SubscribableWebObjectsException . __init__ ( self , "<STR_LIT>" ) <EOL> class InvalidFunctionException ( Exceptions . SubscribableWebObjectsException ) : <EOL> def __init__ ( self ) : <EOL> Exceptions . SubscribableWebObjectsException . __init__ ( self , "<STR_LIT>" ) <EOL> def getObjectClass ( o ) : <EOL> if ComputedGraph . isLocation ( o ) : <EOL> return ComputedGraph . getLocationTypeFromLocation ( o ) <EOL> return o . __class__ <EOL> def calculateLineAndColumn ( message , index ) : <EOL> lineNo = <NUM_LIT:1> <EOL> for line in message . split ( '<STR_LIT:\n>' ) : <EOL> if index < len ( line ) : <EOL> return lineNo , index + <NUM_LIT:1> <EOL> lineNo += <NUM_LIT:1> <EOL> index -= len ( line ) + <NUM_LIT:1> <EOL> def unexpectedExceptionJson ( jsonMessage , message ) : <EOL> return { <EOL> "<STR_LIT>" : jsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:message>" : message <EOL> } <EOL> class IncomingObjectCache ( object ) : <EOL> def __init__ ( self ) : <EOL> self . objectIdCache_ = { } <EOL> self . objectToIdCache_ = { } <EOL> self . objectIdList_ = [ ] <EOL> def lookupObjectById ( self , id ) : <EOL> if id not in self . objectIdCache_ : <EOL> raise MalformedMessageException ( "<STR_LIT>" % id ) <EOL> return self . objectIdCache_ [ id ] <EOL> def addObjectById ( self , id , objectDefinition ) : <EOL> if id in self . objectIdCache_ : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> self . objectIdCache_ [ id ] = objectDefinition <EOL> self . objectIdList_ . append ( id ) <EOL> def flushIdsBelow ( self , threshold ) : <EOL> self . objectIdList_ = sorted ( self . objectIdList_ ) <EOL> while self . objectIdList_ and self . objectIdList_ [ <NUM_LIT:0> ] < threshold : <EOL> del self . objectIdCache_ [ self . objectIdList_ [ <NUM_LIT:0> ] ] <EOL> self . objectIdList_ . pop ( <NUM_LIT:0> ) <EOL> class OutgoingObjectCache ( object ) : <EOL> def __init__ ( self ) : <EOL> self . objectIdCache_ = { } <EOL> self . objectToIdCache_ = { } <EOL> self . curObjectId = <NUM_LIT:0> <EOL> self . maxObjectIdsToKeep = <NUM_LIT> <EOL> self . minObjectId = <NUM_LIT:0> <EOL> def lookupObjectId ( self , obj ) : <EOL> if obj not in self . objectToIdCache_ : <EOL> id = self . newId ( ) <EOL> self . objectToIdCache_ [ obj ] = id <EOL> self . objectIdCache_ [ id ] = obj <EOL> return self . objectToIdCache_ [ obj ] <EOL> def newId ( self ) : <EOL> self . curObjectId += <NUM_LIT:1> <EOL> return self . curObjectId - <NUM_LIT:1> <EOL> def convertResponseToJson ( self , jsonCandidate ) : <EOL> try : <EOL> def raiseException ( ) : <EOL> guid = uuid . uuid4 ( ) <EOL> logging . error ( "<STR_LIT>" , jsonCandidate , type ( jsonCandidate ) , guid ) <EOL> raise Exceptions . SubscribableWebObjectsException ( "<STR_LIT>" % guid ) <EOL> if jsonCandidate is None : <EOL> return jsonCandidate <EOL> if isinstance ( jsonCandidate , ( int , str , unicode , float , bool , long ) ) : <EOL> return jsonCandidate <EOL> if isinstance ( jsonCandidate , ( list , tuple ) ) : <EOL> return [ self . convertResponseToJson ( r ) for r in jsonCandidate ] <EOL> if isinstance ( jsonCandidate , dict ) : <EOL> newDict = { } <EOL> for k , v in jsonCandidate . iteritems ( ) : <EOL> if not isinstance ( k , str ) : <EOL> raiseException ( ) <EOL> newDict [ k ] = self . convertResponseToJson ( v ) <EOL> return newDict <EOL> objDefPopulated = False <EOL> try : <EOL> if ComputedGraph . isLocation ( jsonCandidate ) : <EOL> if jsonCandidate in self . objectToIdCache_ : <EOL> objDef = { <EOL> '<STR_LIT>' : self . lookupObjectId ( jsonCandidate ) <EOL> } <EOL> else : <EOL> objDef = { <EOL> "<STR_LIT>" : self . lookupObjectId ( jsonCandidate ) , <EOL> "<STR_LIT>" : { <EOL> '<STR_LIT:type>' : AllObjectClassesToExpose . typenameFromType ( <EOL> ComputedGraph . getLocationTypeFromLocation ( jsonCandidate ) <EOL> ) , <EOL> '<STR_LIT:args>' : jsonCandidate . __reduce__ ( ) [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> } <EOL> } <EOL> objDefPopulated = True <EOL> else : <EOL> objDef = jsonCandidate . objectDefinition_ <EOL> objDefPopulated = True <EOL> except AttributeError : <EOL> objDefPopulated = False <EOL> if objDefPopulated : <EOL> return self . convertResponseToJson ( objDef ) <EOL> raiseException ( ) <EOL> except : <EOL> logging . error ( "<STR_LIT>" , jsonCandidate , type ( jsonCandidate ) ) <EOL> raise <EOL> def tryFlushObjectIdCache ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . curObjectId > self . minObjectId + self . maxObjectIdsToKeep : <EOL> newMinId = self . curObjectId <EOL> self . minObjectId = newMinId <EOL> self . objectIdCache_ = { } <EOL> self . objectToIdCache_ = { } <EOL> return newMinId <EOL> return None <EOL> def simultaneously ( * factories ) : <EOL> threads = [ ManagedThread . ManagedThread ( target = x ) for x in factories ] <EOL> for t in threads : <EOL> t . start ( ) <EOL> for t in threads : <EOL> t . join ( ) <EOL> class MessageProcessor ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , <EOL> callbackScheduler , <EOL> sharedStateViewFactory , <EOL> computedValueGatewayFactory ) : <EOL> self . lock = threading . Lock ( ) <EOL> self . cacheLoadEvents = { } <EOL> self . resultsById_ = { } <EOL> self . eventsById_ = { } <EOL> logging . info ( "<STR_LIT>" ) <EOL> self . graph = ComputedGraph . ComputedGraph ( ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> Runtime . initialize ( ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> ModuleImporter . initialize ( ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> Fora . _builtin = ForaValue . FORAValue ( ModuleImporter . builtinModuleImplVal ( ) ) <EOL> self . incomingObjectCache = IncomingObjectCache ( ) <EOL> self . outgoingObjectCache = OutgoingObjectCache ( ) <EOL> self . VDM = VectorDataManager . constructVDM ( callbackScheduler ) <EOL> self . VDM . setDropUnreferencedPagesWhenFull ( True ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" , sharedStateViewFactory ) <EOL> def initValueGateway ( ) : <EOL> with self . graph : <EOL> self . computedValueGateway = computedValueGatewayFactory ( ) <EOL> self . cumulusGatewayRemote = self . computedValueGateway . cumulusGateway <EOL> def initSynchronizer ( ) : <EOL> self . synchronizer = SharedStateSynchronizer . SharedStateSynchronizer ( ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> self . synchronizer . attachView ( <EOL> sharedStateViewFactory . createView ( ) <EOL> ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> simultaneously ( <EOL> initSynchronizer , <EOL> initValueGateway <EOL> ) <EOL> self . synchronousSharedStateScope = SynchronousPropertyAccess . SynchronousPropertyAccess ( ) <EOL> self . outstandingMessagesById = { } <EOL> self . expectedMessageId = <NUM_LIT:0> <EOL> self . messageTypeHandlers = { } <EOL> self . messageTypeHandlers [ "<STR_LIT>" ] = self . handleReadMessage <EOL> self . messageTypeHandlers [ "<STR_LIT>" ] = self . handleAssignMessage <EOL> self . messageTypeHandlers [ "<STR_LIT>" ] = self . handleSubscribeMessage <EOL> self . messageTypeHandlers [ "<STR_LIT>" ] = self . handleExecuteMessage <EOL> self . messageTypeHandlers [ "<STR_LIT>" ] = self . handleFlushObjectIds <EOL> self . pendingObjectQueue = [ ] <EOL> self . subscriptions = Subscriptions . Subscriptions ( <EOL> self . graph , <EOL> self . computedValueGateway , <EOL> self . synchronizer <EOL> ) <EOL> def isDisconnectedFromSharedState ( self ) : <EOL> return self . subscriptions . isDisconnectedFromSharedState ( ) <EOL> def handleIncomingMessage ( self , message ) : <EOL> responses = [ ] <EOL> try : <EOL> if message : <EOL> jsonMessage = json . loads ( message , object_hook = Unicode . convertToStringRecursively ) <EOL> responses = self . handleJsonMessage ( jsonMessage ) <EOL> responses += self . tryFlushObjectIdCache ( ) <EOL> else : <EOL> responses = self . updateGraphAndReturnMessages ( ) <EOL> except UnicodeEncodeError as err : <EOL> self . expectedMessageId += <NUM_LIT:1> <EOL> jsonRequest = json . loads ( message ) <EOL> line , column = calculateLineAndColumn ( err . object , err . start ) <EOL> what = { <EOL> '<STR_LIT>' : line , <EOL> '<STR_LIT>' : column , <EOL> '<STR_LIT>' : err . object [ err . start : err . end ] <EOL> } <EOL> responses = [ unexpectedExceptionJson ( jsonRequest , what ) ] <EOL> return responses <EOL> def handleJsonMessage ( self , incomingJsonMessage ) : <EOL> if not isinstance ( incomingJsonMessage , dict ) : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" % incomingJsonMessage ) <EOL> if not '<STR_LIT>' in incomingJsonMessage : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" % incomingJsonMessage ) <EOL> if incomingJsonMessage [ '<STR_LIT>' ] != self . expectedMessageId : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" % <EOL> ( self . expectedMessageId , incomingJsonMessage [ '<STR_LIT>' ] , incomingJsonMessage ) <EOL> ) <EOL> try : <EOL> self . expectedMessageId += <NUM_LIT:1> <EOL> if incomingJsonMessage [ '<STR_LIT>' ] not in self . messageTypeHandlers : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> if not '<STR_LIT>' in incomingJsonMessage : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> if incomingJsonMessage [ "<STR_LIT>" ] != "<STR_LIT>" : <EOL> obj = self . extractObjectDefinition ( incomingJsonMessage [ '<STR_LIT>' ] ) <EOL> else : <EOL> obj = None <EOL> return self . messageTypeHandlers [ incomingJsonMessage [ "<STR_LIT>" ] ] ( incomingJsonMessage , obj ) <EOL> except MalformedMessageException : <EOL> raise <EOL> except Exception as e : <EOL> return [ unexpectedExceptionJson ( incomingJsonMessage , Exceptions . wrapException ( e ) . message ) ] <EOL> def tryFlushObjectIdCache ( self ) : <EOL> objectId = self . outgoingObjectCache . tryFlushObjectIdCache ( ) <EOL> if objectId is not None : <EOL> return [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:value>' : objectId } ] <EOL> return [ ] <EOL> def handleFlushObjectIds ( self , incomingJsonMessage , object ) : <EOL> if not '<STR_LIT:value>' in incomingJsonMessage : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> threshold = incomingJsonMessage [ '<STR_LIT:value>' ] <EOL> if not isinstance ( threshold , int ) : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> self . incomingObjectCache . flushIdsBelow ( threshold ) <EOL> return [ { <EOL> "<STR_LIT>" : incomingJsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT:OK>" <EOL> } ] <EOL> def extractObjectDefinition ( self , objDefJson ) : <EOL> if '<STR_LIT>' in objDefJson or '<STR_LIT>' in objDefJson : <EOL> return self . convertObjectArgs ( objDefJson ) <EOL> if '<STR_LIT:type>' not in objDefJson or '<STR_LIT:args>' not in objDefJson : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> objType = objDefJson [ '<STR_LIT:type>' ] <EOL> objectArgs = objDefJson [ '<STR_LIT:args>' ] <EOL> objectArgs = self . convertObjectArgs ( objectArgs ) <EOL> if objType not in AllObjectClassesToExpose . classMap : <EOL> raise InvalidObjectDefinitionException ( "<STR_LIT>" ) <EOL> try : <EOL> objectCls = AllObjectClassesToExpose . classMap [ objType ] <EOL> result = objectCls ( objectArgs ) <EOL> if not ComputedGraph . isLocation ( result ) : <EOL> result . __dict__ [ '<STR_LIT>' ] = objDefJson <EOL> return result <EOL> except Exceptions . SubscribableWebObjectsException as e : <EOL> raise InvalidObjectDefinitionException ( e . message ) <EOL> def convertObjectArgs ( self , objectArgs ) : <EOL> if isinstance ( objectArgs , list ) : <EOL> return [ self . convertObjectArgs ( x ) for x in objectArgs ] <EOL> if isinstance ( objectArgs , dict ) : <EOL> if '<STR_LIT>' in objectArgs : <EOL> obj = self . extractObjectDefinition ( objectArgs [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in objectArgs : <EOL> self . incomingObjectCache . addObjectById ( objectArgs [ '<STR_LIT>' ] , obj ) <EOL> return obj <EOL> if '<STR_LIT>' in objectArgs : <EOL> obj = self . incomingObjectCache . lookupObjectById ( objectArgs [ '<STR_LIT>' ] ) <EOL> return obj <EOL> tr = { } <EOL> for k , v in objectArgs . iteritems ( ) : <EOL> tr [ k ] = self . convertObjectArgs ( v ) <EOL> return tr <EOL> return objectArgs <EOL> def getFieldExtractorForReadMessage ( self , jsonMessage , objectToRead ) : <EOL> if '<STR_LIT>' not in jsonMessage : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> field = jsonMessage [ '<STR_LIT>' ] <EOL> if not isinstance ( field , str ) : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> field = intern ( field ) <EOL> try : <EOL> fieldDef = getattr ( getObjectClass ( objectToRead ) , field ) <EOL> except : <EOL> raise InvalidFieldException ( ) <EOL> if not Decorators . isPropertyToExpose ( fieldDef ) : <EOL> raise InvalidFieldException ( ) <EOL> def getter ( x ) : <EOL> return getattr ( x , field ) <EOL> return getter <EOL> def getFunctionForExecuteMessage ( self , jsonMessage , objectToRead ) : <EOL> if '<STR_LIT>' not in jsonMessage : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> field = jsonMessage [ '<STR_LIT>' ] <EOL> if not isinstance ( field , str ) : <EOL> raise MalformedMessageException ( "<STR_LIT>" ) <EOL> field = intern ( field ) <EOL> try : <EOL> funDef = getattr ( getObjectClass ( objectToRead ) , field ) <EOL> except : <EOL> raise InvalidFieldException ( ) <EOL> funDef = getattr ( getObjectClass ( objectToRead ) , field ) <EOL> if not Decorators . isFunctionToExpose ( funDef ) : <EOL> raise InvalidFunctionException ( ) <EOL> if Decorators . functionExpectsCallback ( funDef ) : <EOL> if Decorators . functionExpectsExpandedArgs ( funDef ) : <EOL> def caller ( x , callback , arg ) : <EOL> if isinstance ( arg , list ) : <EOL> return getattr ( x , field ) ( callback , * arg ) <EOL> else : <EOL> return getattr ( x , field ) ( callback , ** arg ) <EOL> return caller <EOL> else : <EOL> return lambda x , callback , * args : getattr ( x , field ) ( callback , * args ) <EOL> else : <EOL> if Decorators . functionExpectsExpandedArgs ( funDef ) : <EOL> def caller ( x , callback , arg ) : <EOL> try : <EOL> if isinstance ( arg , list ) : <EOL> callback ( getattr ( x , field ) ( * arg ) ) <EOL> else : <EOL> callback ( getattr ( x , field ) ( ** arg ) ) <EOL> except Exception as e : <EOL> callback ( e ) <EOL> return caller <EOL> else : <EOL> def caller ( x , callback , * args ) : <EOL> try : <EOL> callback ( getattr ( x , field ) ( * args ) ) <EOL> except Exception as e : <EOL> callback ( e ) <EOL> return caller <EOL> def addResultToQueue ( self , messageCreator ) : <EOL> self . pendingObjectQueue . append ( messageCreator ) <EOL> def handleExecuteMessage ( self , jsonMessage , objectToExecuteOn ) : <EOL> funImplementation = self . getFunctionForExecuteMessage ( jsonMessage , objectToExecuteOn ) <EOL> def callback ( result ) : <EOL> if isinstance ( result , Exception ) : <EOL> def creator ( ) : <EOL> return unexpectedExceptionJson ( jsonMessage , Exceptions . wrapException ( result ) . message ) <EOL> self . addResultToQueue ( creator ) <EOL> else : <EOL> def creator ( ) : <EOL> try : <EOL> return { <EOL> "<STR_LIT>" : jsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:result>" : self . outgoingObjectCache . convertResponseToJson ( result ) <EOL> } <EOL> except Exception as e : <EOL> return unexpectedExceptionJson ( jsonMessage , Exceptions . wrapException ( e ) . message ) <EOL> self . addResultToQueue ( creator ) <EOL> try : <EOL> objectArgs = self . convertObjectArgs ( jsonMessage [ '<STR_LIT:args>' ] ) <EOL> funImplementation ( objectToExecuteOn , callback , objectArgs ) <EOL> return [ ] <EOL> except Exception as e : <EOL> return [ unexpectedExceptionJson ( jsonMessage , Exceptions . wrapException ( e ) . message ) ] <EOL> def handleReadMessage ( self , jsonMessage , objectToRead ) : <EOL> fieldFun = self . getFieldExtractorForReadMessage ( jsonMessage , objectToRead ) <EOL> try : <EOL> return [ { <EOL> "<STR_LIT>" : jsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : self . outgoingObjectCache . convertResponseToJson ( fieldFun ( objectToRead ) ) <EOL> } ] <EOL> except Exception as e : <EOL> try : <EOL> objectType = objectToRead . __location_class__ if ComputedGraph . isLocation ( objectToRead ) else str ( objectToRead ) <EOL> logging . info ( "<STR_LIT>" , fieldFun ( objectToRead ) , objectType ) <EOL> except : <EOL> pass <EOL> return [ unexpectedExceptionJson ( jsonMessage , Exceptions . wrapException ( e ) . message ) ] <EOL> def handleSubscribeMessage ( self , jsonMessage , objectToRead ) : <EOL> fieldFun = self . getFieldExtractorForReadMessage ( jsonMessage , objectToRead ) <EOL> getValueFun = lambda : fieldFun ( objectToRead ) <EOL> value , otherSubscriptionChanges = self . subscriptions . addSubscription ( <EOL> jsonMessage [ '<STR_LIT>' ] , <EOL> getValueFun <EOL> ) <EOL> if isinstance ( value , Exceptions . SubscribableWebObjectsException ) : <EOL> valueResponseJson = unexpectedExceptionJson ( jsonMessage , value . message ) <EOL> else : <EOL> valueResponseJson = { <EOL> "<STR_LIT>" : jsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : self . outgoingObjectCache . convertResponseToJson ( fieldFun ( objectToRead ) ) <EOL> } <EOL> return ( [ valueResponseJson ] + <EOL> self . encodeSubscriptionChangesAsJsonAndDrop ( otherSubscriptionChanges ) <EOL> ) <EOL> def encodeSubscriptionChangesAsJsonAndDrop ( self , changedSubscriptionIds ) : <EOL> result = [ ] <EOL> for messageId in changedSubscriptionIds : <EOL> value = self . subscriptions . getValueAndDropSubscription ( messageId ) <EOL> try : <EOL> if isinstance ( value , Exceptions . SubscribableWebObjectsException ) : <EOL> valueResponseJson = { <EOL> "<STR_LIT>" : messageId , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:message>" : value . message <EOL> } <EOL> else : <EOL> valueResponseJson = { <EOL> "<STR_LIT>" : messageId , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:value>" : self . outgoingObjectCache . convertResponseToJson ( value ) <EOL> } <EOL> except : <EOL> guid = uuid . uuid4 ( ) <EOL> logging . error ( <EOL> "<STR_LIT>" , <EOL> guid , <EOL> traceback . format_exc ( ) <EOL> ) <EOL> valueResponseJson = { <EOL> "<STR_LIT>" : messageId , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:message>" : "<STR_LIT>" % guid <EOL> } <EOL> result . append ( valueResponseJson ) <EOL> return result <EOL> def updateGraphAndReturnMessages ( self ) : <EOL> changed = self . subscriptions . updateAndReturnChangedSubscriptionIds ( ) <EOL> return self . encodeSubscriptionChangesAsJsonAndDrop ( changed ) <EOL> def extractPendingMessages ( self ) : <EOL> tr = [ x ( ) for x in self . pendingObjectQueue ] <EOL> self . pendingObjectQueue = [ ] <EOL> return tr <EOL> def handleAssignMessage ( self , jsonMessage , objectToRead ) : <EOL> if '<STR_LIT>' not in jsonMessage : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" + str ( jsonMessage ) <EOL> ) <EOL> if '<STR_LIT:value>' not in jsonMessage : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" + str ( jsonMessage ) <EOL> ) <EOL> field = jsonMessage [ '<STR_LIT>' ] <EOL> if not isinstance ( field , str ) : <EOL> raise MalformedMessageException ( <EOL> "<STR_LIT>" + str ( jsonMessage ) ) <EOL> field = intern ( field ) <EOL> try : <EOL> fieldDef = getattr ( getObjectClass ( objectToRead ) , field ) <EOL> except : <EOL> raise InvalidFieldException ( ) <EOL> fieldDef = getattr ( getObjectClass ( objectToRead ) , field ) <EOL> if not Decorators . isPropertyToExpose ( fieldDef ) : <EOL> raise InvalidFieldException ( ) <EOL> if not Decorators . propertyHasSetter ( fieldDef ) : <EOL> return InvalidFieldException ( ) <EOL> try : <EOL> setter = Decorators . getSetter ( field , fieldDef ) <EOL> setter ( objectToRead , jsonMessage [ '<STR_LIT:value>' ] ) <EOL> return [ { <EOL> "<STR_LIT>" : jsonMessage [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT:OK>" <EOL> } ] <EOL> except Exception as e : <EOL> return [ unexpectedExceptionJson ( jsonMessage , Exceptions . wrapException ( e ) . message ) ] <EOL> def __enter__ ( self ) : <EOL> self . lock . __enter__ ( ) <EOL> self . synchronizer . __enter__ ( ) <EOL> self . computedValueGateway . __enter__ ( ) <EOL> self . graph . __enter__ ( ) <EOL> self . synchronousSharedStateScope . __enter__ ( ) <EOL> def __exit__ ( self , type , value , tb ) : <EOL> self . synchronousSharedStateScope . __exit__ ( type , value , tb ) <EOL> self . graph . __exit__ ( type , value , tb ) <EOL> self . computedValueGateway . __exit__ ( type , value , tb ) <EOL> self . synchronizer . __exit__ ( type , value , tb ) <EOL> self . lock . __exit__ ( type , value , tb ) <EOL> def teardown ( self ) : <EOL> self . synchronizer . flush ( ) <EOL> self . computedValueGateway . teardown ( ) <EOL> self . synchronizer = None <EOL> self . graph = None <EOL> self . computedValueGateway = None </s>
