Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return None, None
def describe(self):
if self.request is None:
return ""
return "" % self.request.reqID
def truncate(s, limit):
assert limit > 3
if s and len(s) > limit:
s = s[:limit-3] + ".."
return s
# failures are sent as Copyables
class FailureSlicer(slicer.BaseSlicer):
slices = failure.Failure
classname = "twisted.python.failure.Failure"
def slice(self, streamable, banana):
self.streamable = streamable
yield 'copyable'
yield self.classname
state = self.getStateToCopy(self.obj, banana)
for k,v in state.items():
yield k
yield v
def describe(self):
return "<%s>" % self.classname
def getStateToCopy(self, obj, broker):
#state = obj.__dict__.copy()
# -*- test-case-name: foolscap.test.test_banana -*-
from twisted.python.components import registerAdapter
from twisted.internet.defer import Deferred
from foolscap import tokens
from foolscap.tokens import Violation, BananaError
from foolscap.slicer import BaseSlicer, LeafUnslicer
from foolscap.constraint import OpenerConstraint, IntegerConstraint, Any
class BooleanSlicer(BaseSlicer):
opentype = ('boolean',)
trackReferences = False
def sliceBody(self, streamable, banana):
if self.obj:
yield 1
else:
yield 0
registerAdapter(BooleanSlicer, bool, tokens.ISlicer)
class BooleanUnslicer(LeafUnslicer):
opentype = ('boolean',)
value = None
constraint = None
def setConstraint(self, constraint):
meth = getattr(self.original, "remote_" + methname)
return meth(*args, **kwargs)
d = fireEventually()
d.addCallback(_try)
return d
def callRemoteOnly(self, methname, *args, **kwargs):
d = self.callRemote(methname, *args, **kwargs)
d.addErrback(lambda f: None)
return None
registerAdapter(LocalReferenceable, ipb.IReferenceable, ipb.IRemoteReference)
class YourReferenceSlicer(slicer.BaseSlicer):
"""I handle pb.RemoteReference objects (being sent back home to the
original pb.Referenceable-holder)
"""
def slice(self, streamable, protocol):
broker = self.requireBroker(protocol)
self.streamable = streamable
tracker = self.obj.tracker
if tracker.broker == broker:
# sending back to home broker
yield b'your-reference'
yield tracker.clid
else:
# sending somewhere else
furl = tracker.getURL()
if furl is None:
# their-reference sequences, to prompt the eventual recipient to
# create a new connection for this object.
# a big note on object lifetimes: obviously, the data cannot keep
# the Referenceable alive. Use tub.registerReference() on any
# Referenceable that you want to include in the serialized data,
# and take steps to make sure that later incarnations of this Tub
# will do the same.
yield b'their-reference'
yield 0 # giftID==0 tells the recipient to not try to ack it
yield six.ensure_binary(tracker.getURL())
registerAdapter(ReferenceableSlicer, Referenceable, ipb.ISlicer)
class CallableSlicer(slicer.BaseSlicer):
"""Bound methods are serialized as my-reference sequences with negative
clid values."""
opentype = ('my-reference',)
def sliceBody(self, streamable, protocol):
broker = self.requireBroker(protocol)
# TODO: consider this requirement, maybe based upon a Tub flag
# assert ipb.ISlicer(self.obj.im_self)
# or maybe even isinstance(self.obj.im_self, Referenceable)
puid = id(self.obj)
tracker = broker.getTrackerForMyCall(puid, self.obj)
yield tracker.clid
firstTime = tracker.send()
if firstTime:
# this is the first time the Call has crossed this wire. In
# addition to the clid, send the schema name and any URL this
class ModuleSlicer(slicer.BaseSlicer):
opentype = ('module',)
trackReferences = True
def sliceBody(self, streamable, banana):
yield self.obj.__name__
class ClassSlicer(slicer.BaseSlicer):
opentype = ('class',)
trackReferences = True
def sliceBody(self, streamable, banana):
yield reflect.qual(self.obj)
class MethodSlicer(slicer.BaseSlicer):
opentype = ('method',)
trackReferences = True
def sliceBody(self, streamable, banana):
yield self.obj.im_func.__name__
yield self.obj.im_self
yield self.obj.im_class
class FunctionSlicer(slicer.BaseSlicer):
opentype = ('function',)
trackReferences = True
def sliceBody(self, streamable, banana):
name = self.obj.__name__
fullname = str(whichmodule(self.obj, self.obj.__name__)) + '.' + name
yield fullname
opentype = ('class',)
trackReferences = True
def sliceBody(self, streamable, banana):
yield reflect.qual(self.obj)
class MethodSlicer(slicer.BaseSlicer):
opentype = ('method',)
trackReferences = True
def sliceBody(self, streamable, banana):
yield self.obj.im_func.__name__
yield self.obj.im_self
yield self.obj.im_class
class FunctionSlicer(slicer.BaseSlicer):
opentype = ('function',)
trackReferences = True
def sliceBody(self, streamable, banana):
name = self.obj.__name__
fullname = str(whichmodule(self.obj, self.obj.__name__)) + '.' + name
yield fullname
UnsafeSlicerTable = {}
UnsafeSlicerTable.update({
types.InstanceType: InstanceSlicer,
types.ModuleType: ModuleSlicer,
types.ClassType: ClassSlicer,
types.MethodType: MethodSlicer,
types.FunctionType: FunctionSlicer,
#types.TypeType: NewstyleClassSlicer,
# -*- test-case-name: foolscap.test.test_banana -*-
import six
import re
from twisted.internet.defer import Deferred
from foolscap.tokens import BananaError, STRING, VOCAB, Violation
from foolscap.slicer import BaseSlicer, LeafUnslicer
from foolscap.constraint import OpenerConstraint, Any
class UnicodeSlicer(BaseSlicer):
opentype = ("unicode",)
slices = six.text_type
def sliceBody(self, streamable, banana):
yield self.obj.encode("UTF-8")
class UnicodeUnslicer(LeafUnslicer):
# accept a UTF-8 encoded string
opentype = ("unicode",)
string = None
constraint = None
def setConstraint(self, constraint):
if isinstance(constraint, Any):
return
assert isinstance(constraint, UnicodeConstraint)
self.constraint = constraint
# -*- test-case-name: foolscap.test.test_banana -*-
import decimal
import six
from twisted.internet.defer import Deferred
from foolscap.tokens import BananaError, STRING, VOCAB
from foolscap.slicer import BaseSlicer, LeafUnslicer
from foolscap.constraint import Any
class DecimalSlicer(BaseSlicer):
opentype = ("decimal",)
slices = decimal.Decimal
def sliceBody(self, streamable, banana):
yield six.ensure_binary(str(self.obj))
class DecimalUnslicer(LeafUnslicer):
opentype = ("decimal",)
value = None
constraint = None
def setConstraint(self, constraint):
if isinstance(constraint, Any):
return
assert False, "DecimalUnslicer does not currently accept a constraint"
def checkToken(self, typebyte, size):
# -*- test-case-name: foolscap.test.test_banana -*-
from twisted.python import log
from twisted.internet.defer import Deferred
from foolscap.tokens import Violation, BananaError
from foolscap.slicer import BaseSlicer, BaseUnslicer
from foolscap.constraint import OpenerConstraint, Any, IConstraint
from foolscap.util import AsyncAND
class DictSlicer(BaseSlicer):
opentype = ('dict',)
trackReferences = True
slices = None
def sliceBody(self, streamable, banana):
for key,value in self.obj.items():
yield key
yield value
class DictUnslicer(BaseUnslicer):
opentype = ('dict',)
gettingKey = True
keyConstraint = None
valueConstraint = None
maxKeys = None
def receiveClose(self):
if self.key is not None:
raise BananaError("sequence ended early: got key but not value")
# now is the time we replace our protocol's vocab table
self.protocol.replaceIncomingVocabulary(self.d)
return ReplaceVocabularyTable, None
def describe(self):
if self.key is not None:
return "[%s]" % self.key
else:
return ""
class AddVocabSlicer(BaseSlicer):
opentype = ('add-vocab',)
trackReferences = False
def __init__(self, value):
assert isinstance(value, str)
self.value = value
def slice(self, streamable, banana):
# we need to implement slice() (instead of merely sliceBody) so we
# can get control at the beginning and end of serialization. It also
# gives us access to the Banana protocol object, so we can manipulate
# their outgoingVocabulary table.
self.streamable = streamable
self.start(banana)
for o in self.opentype:
yield o