<commit_before># coding: utf-8
from twisted.python import log
from twisted.internet import defer, protocol
from twisted.protocols import basic
from qpid import spec, codec, connection
from qpid.message import Message
from qpid.content import Content
from qpid.peer import Closed
from StringIO import StringIO
import struct

class GarbageException(Exception):
    pass

# An AMQP channel is a virtual connection that shares the
# same socket with others channels. One can have many channels
# per connection
class AMQChannel(object):

    def __init__(self, id, proto, spec):
        self.id = id
        self.proto = proto
        self.spec = spec
        self.deferred = defer.Deferred()
        self.incoming = defer.DeferredQueue()
        self.worker()
        self.closed = False
        self.reason = None

    def _dispatchContent(self, content, payload):
        message = Message(payload.method, payload.args, content)
        self.messageReceived(message)

    def worker(self, frame=None):
        if frame:
            payload = frame.payload
            if payload.method.content:
                d = readContent(self.incoming)
            else:
                d = defer.succeed(None)
            d.addCallback(self._dispatchContent, payload).addCallback(self.worker)
        else:
            self.incoming.get().addCallback(self.worker)

    def sendFrame(self, frame):
        self.proto.sendFrame(frame)

        payload = frame.payload
        if isinstance(payload, connection.Method):
            if len(payload.method.responses) > 0:
                self.deferred = defer.Deferred()
        return self.deferred
 
    def close(self, reason):
        self.closed = True
        self.reason = reason
        raise Closed(self.reason)

    def invoke(self, method, args, content=None):
        if self.closed:
            raise Closed(self.reason)
        frame = connection.Frame(self.id, connection.Method(method, *args))

        d = self.sendFrame(frame)

        if method.content:
          if content is None:
            content = Content()
          self.writeContent(method.klass, content)

        return d

    def writeContent(self, klass, content):
        size = content.size()
        header = connection.Frame(self.id, connection.Header(klass, content.weight(), size, **content.properties))
        self.sendFrame(header)
        for child in content.children:
            self.writeContent(klass, child)
        # should split up if content.body exceeds max frame size
        if size > 0:
            self.sendFrame(connection.Frame(self.id, connection.Body(content.body)))

    def frameReceived(self, frame):
        self.incoming.put(frame)
 
    def messageReceived(self, message):
        deferred, self.deferred = self.deferred, defer.Deferred()
        deferred.callback(message)
        return deferred

class FrameReceiver(protocol.Protocol, basic._PauseableMixin):

    frame_mode = False
    MAX_LENGTH = 16384
    __buffer = ''

    def __init__(self, spec, *args, **kwargs):
        self.spec = spec
        self.FRAME_END = self.spec.constants.bypyname["frame_end"].id

    # packs a frame and writes it to the underlying transport
    def sendFrame(self, frame):
        data = self._packFrame(frame)
        self.transport.write(data)

    # packs a frame, see qpid.connection.Connection#write
    def _packFrame(self, frame):
        s = StringIO()
        c = codec.Codec(s)
        c.encode_octet(self.spec.constants.bypyname[frame.payload.type].id)
        c.encode_short(frame.channel)
        frame.payload.encode(c)
        c.encode_octet(self.FRAME_END)
        data = s.getvalue()
        return data

    # packs a frame, see qpid.connection.Connection#read
    def _unpackFrame(self, data):
        try:
            s = StringIO(data)
            c = codec.Codec(s)
            frameType = spec.pythonize(self.spec.constants.byid[c.decode_octet()].name)
            channel = c.decode_short()
            payload = connection.Frame.DECODERS[frameType].decode(self.spec, c)
            end = c.decode_octet()
            if end != self.FRAME_END:
                raise GarbageException('frame error: expected %r, got %r' % (self.FRAME_END, end))
            frame = connection.Frame(channel, payload)
            return frame, data[s.pos:]
        except codec.EOF:
            return data
        except struct.error:<commit_msg>use cStringIO instead of StringIO<commit_after># coding: utf-8
from twisted.python import log
from twisted.internet import defer, protocol
from twisted.protocols import basic
from qpid import spec, codec, connection
from qpid.message import Message
from qpid.content import Content
from qpid.peer import Closed
from cStringIO import StringIO
import struct

class GarbageException(Exception):
    pass

# An AMQP channel is a virtual connection that shares the
# same socket with others channels. One can have many channels
# per connection
class AMQChannel(object):

    def __init__(self, id, proto, spec):
        self.id = id
        self.proto = proto
        self.spec = spec
        self.deferred = defer.Deferred()
        self.incoming = defer.DeferredQueue()
        self.worker()
        self.closed = False
        self.reason = None

    def _dispatchContent(self, content, payload):
        message = Message(payload.method, payload.args, content)
        self.messageReceived(message)

    def worker(self, frame=None):
        if frame:
            payload = frame.payload
            if payload.method.content:
                d = readContent(self.incoming)
            else:
                d = defer.succeed(None)
            d.addCallback(self._dispatchContent, payload).addCallback(self.worker)
        else:
            self.incoming.get().addCallback(self.worker)

    def sendFrame(self, frame):
        self.proto.sendFrame(frame)

        payload = frame.payload
        if isinstance(payload, connection.Method):
            if len(payload.method.responses) > 0:
                self.deferred = defer.Deferred()
        return self.deferred
 
    def close(self, reason):
        self.closed = True
        self.reason = reason
        raise Closed(self.reason)

    def invoke(self, method, args, content=None):
        if self.closed:
            raise Closed(self.reason)
        frame = connection.Frame(self.id, connection.Method(method, *args))

        d = self.sendFrame(frame)

        if method.content:
          if content is None:
            content = Content()
          self.writeContent(method.klass, content)

        return d

    def writeContent(self, klass, content):
        size = content.size()
        header = connection.Frame(self.id, connection.Header(klass, content.weight(), size, **content.properties))
        self.sendFrame(header)
        for child in content.children:
            self.writeContent(klass, child)
        # should split up if content.body exceeds max frame size
        if size > 0:
            self.sendFrame(connection.Frame(self.id, connection.Body(content.body)))

    def frameReceived(self, frame):
        self.incoming.put(frame)
 
    def messageReceived(self, message):
        deferred, self.deferred = self.deferred, defer.Deferred()
        deferred.callback(message)
        return deferred

class FrameReceiver(protocol.Protocol, basic._PauseableMixin):

    frame_mode = False
    MAX_LENGTH = 16384
    __buffer = ''

    def __init__(self, spec, *args, **kwargs):
        self.spec = spec
        self.FRAME_END = self.spec.constants.bypyname["frame_end"].id

    # packs a frame and writes it to the underlying transport
    def sendFrame(self, frame):
        data = self._packFrame(frame)
        self.transport.write(data)

    # packs a frame, see qpid.connection.Connection#write
    def _packFrame(self, frame):
        s = StringIO()
        c = codec.Codec(s)
        c.encode_octet(self.spec.constants.bypyname[frame.payload.type].id)
        c.encode_short(frame.channel)
        frame.payload.encode(c)
        c.encode_octet(self.FRAME_END)
        data = s.getvalue()
        return data

    # packs a frame, see qpid.connection.Connection#read
    def _unpackFrame(self, data):
        try:
            s = StringIO(data)
            c = codec.Codec(s)
            frameType = spec.pythonize(self.spec.constants.byid[c.decode_octet()].name)
            channel = c.decode_short()
            payload = connection.Frame.DECODERS[frameType].decode(self.spec, c)
            end = c.decode_octet()
            if end != self.FRAME_END:
                raise GarbageException('frame error: expected %r, got %r' % (self.FRAME_END, end))
            frame = connection.Frame(channel, payload)
            return frame, s.read()
        except codec.EOF:
            return data
        except struct.error:
{'fraction_non_alphanumeric': 0.09640059043942319, 'fraction_numerical': 0.0018167366867264676, 'mean_word_length': 2.3708381171067736, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '496606', 'n_tokens_mistral': 2504, 'n_tokens_neox': 2428, 'n_words': 627}