def sendMessage(self, msg, stream, whenDone=None):
        """
        Queues L{msg} for encoding into RTMP and being spit out to L{output}.

        This is a low level api and is not for public consumption. Use
        L{interfaces.IStream.sendMessage} instead.

        @param msg: The message being sent to the peer.
        @type msg: L{message.IMessage}
        @param stream: The stream instance that is sending the message.
        @type stream: L{interfaces.IStream}
        @param whenDone: A callback fired when the message has been written to
            the RTMP stream. See L{BaseStream.sendMessage}
        """
        buf = BufferedByteStream()
        e = self.encoder

        # this will probably need to be rethought as this could block for an
        # unacceptable amount of time. For most messages however it seems to be
        # fast enough and the penalty for setting up a new thread is too high.
        msg.encode(buf)

        e.send(buf.getvalue(), msg.__data_type__,
            stream.streamId, stream.timestamp, whenDone)

        if e.active and not self._encoding_task:
            self.startEncoding()

    def sendMessage(self, msg, stream, whenDone=None):
        """
        Sends an RTMP message to the peer. Not part of a public api, use
        C{stream.sendMessage} instead.

        @param msg: The message being sent to the peer.
        @type msg: L{message.IMessage}
        @param stream: The stream instance that is sending the message.
        @type stream: L{NetStream}
        @param whenDone: A callback fired when the message has been written to
            the RTMP stream. See L{BaseStream.sendMessage}
        """
        buf = BufferedByteStream()
        e = self.encoder

        # this will probably need to be rethought as this could block for an
        # unacceptable amount of time. For most messages however it seems to be
        # fast enough and the penalty for setting up a new thread is too high.
        msg.encode(buf)

        e.send(buf.getvalue(), msg.__data_type__,
            stream.streamId, stream.timestamp, whenDone)

        if e.active and not self.encoder_task:
            self.startEncoding()

    def test_assemble(self):
        for fixture in self.data:
            buf = BufferedByteStream()
            a = RTMPAssembler(128, buf)

            for packet in fixture['packets']:
                a.push_packet(packet)

            buf.seek(0, 0)

            self.failUnlessEqual(struct.pack("B" * len(fixture['data']), *fixture['data']), buf.read())

    def test_assemble_chunks(self):
        chunkSizes = (32, 64, 128, 256)
        header = RTMPHeader(object_id=2, timestamp=9504486, length=10, type=0x04, stream_id=0)

        for chunkSize in chunkSizes:
            for l in xrange(1, 258):
                data = ''.join([chr(random.randint(0, 255)) for x in xrange(l)])

                buf = BufferedByteStream()
                a = RTMPAssembler(chunkSize, buf)
                a.push_packet(DataPacket(header=header, data=data))

                buf.seek(0, 0)
                self.failUnlessEqual(''.join(self.gen_packet("\x02\x91\x06\xe6\x00\x00\x01\x04\x00\x00\x00\x00", ["\xc2"], data, l, chunkSize)), buf.read())

    def startStreaming(self):
        """
        This must be called before any RTMP data is received.
        """
        self.streamManager = self.buildStreamManager()
        self.controlStream = self.streamManager.getControlStream()

        self._decodingBuffer = BufferedByteStream()
        self._encodingBuffer = BufferedByteStream()

        self.decoder = codec.Decoder(self.getDispatcher(), self.streamManager, stream=self._decodingBuffer)
        self.encoder = codec.Encoder(self.getWriter(), stream=self._encodingBuffer)

        self.decoder_task = None
        self.encoder_task = None

    class _BackRelay(protocol.ProcessProtocol):
        def __init__(self, deferred):
            self.deferred = deferred
            self.s = BufferedByteStream()

        def errReceived(self, text):
            self.deferred.errback(failure.Failure(IOError()))
            self.deferred = None
            self.transport.loseConnection()

        def outReceived(self, text):
            self.s.write(text)

        def processEnded(self, reason):
            if self.deferred is not None:
                result = self.s.getvalue()
                self.deferred.callback(result)

    def connectionMade(self):
        """
        Successfully connected to peer.
        """
        self.input = RTMPDisassembler(constants.DEFAULT_CHUNK_SIZE)
        self.output = RTMPAssembler(constants.DEFAULT_CHUNK_SIZE, self.transport)

        self.state = self.State.HANDSHAKE_SEND
        self.handshakeTimeout = reactor.callLater(config.getint('RTMP', 'handshakeTimeout'), self._handshakeTimedout)
        self.handshakeBuf = BufferedByteStream()
        self._beginHandshake()

    def __init__(self, chunkSize):
        """
        Constructor.

        @param chunkSize: initial size of chunk
        @type chunkSize: C{int}
        """
        self.lastHeaders = {}
        self.pool = {}
        self.chunkSize = chunkSize
        self.buffer = BufferedByteStream()

    def start(self, uptime=0, version=0):
        """
        Called to start the handshaking negotiations.
        """
        if self.started:
            raise AlreadyStarted('Handshake negotiator cannot be restarted')

        self.started = True

        self.uptime = uptime
        self.version = version

        self._buffer = BufferedByteStream()

    def write(self):
        """
        Encode packet into bytes.

        @return: representation of packet
        @rtype: C{str}
        """
        buf = BufferedByteStream()
        buf.write_ushort(self.event)

        for val in self.data:
            buf.write_ulong(val)

        self.header.length = len(buf)
        buf.seek(0, 0)
        return buf.read()

class ProducingChannel(BaseChannel):
    """
    Writes RTMP frames.

    @ivar buffer: Any data waiting to be written to the underlying stream.
    @type buffer: L{BufferedByteStream}
    @ivar acquired: Whether this channel is acquired. See L{ChannelMuxer.
        acquireChannel}
    """


    def __init__(self, channelId, stream, frameSize):
        BaseChannel.__init__(self, channelId, stream, frameSize)

        self.buffer = BufferedByteStream()
        self.acquired = False
        self.callback = None


    def setCallback(self, cb):
        """
        Sets the callback that will be fired once this channel has been completely
        encoded.
        """
        self.callback = cb


    def reset(self):
        """
        Called when the channel has completed writing the buffer.
        """
        BaseChannel.reset(self)

        self.buffer.seek(0)
        self.buffer.truncate()
        self.header = None


    def append(self, data):
        """
        Appends data to the buffer in preparation of encoding in RTMP.
        """
        self.buffer.append(data)


    def marshallFrame(self, size):
        """
        Writes a section of the buffer as part of the RTMP frame.
        """
        self.stream.write(self.buffer.read(size))

    def __init__(self, channel, streamId, output):
        self.type = None
        self.channel = channel
        self.streamId = streamId
        self.output = output
        self.stream = BufferedByteStream()

        self._lastHeader = None
        self._oldStream = channel.stream
        channel.stream = self.stream

        h = header.Header(channel.channelId)

        # encode a continuation header for speed
        header.encode(self.stream, h, h)

        self._continuationHeader = self.stream.getvalue()
        self.stream.consume()

    def start(self, uptime=None, version=None):
        """
        Called to start the handshaking negotiations.
        """
        if self.started:
            raise HandshakeError('Handshake negotiator cannot be restarted')

        self.started = True
        self.buffer = BufferedByteStream()

        self.peer_version = None

        self.my_syn = Packet(uptime, version)
        self.my_ack = None

        self.peer_syn = None
        self.peer_ack = None

        self.buildSynPayload(self.my_syn)

        self._writePacket(self.my_syn)

 def __init__(self):
     self.buffer = BufferedByteStream()

    def __init__(self):
        self.buffer = BufferedByteStream()

        self.channels = {}
        self.frameSize = FRAME_SIZE
        self.bytes = 0

class Packet(object):
    """
    """


    format = None
    challengeKey = None


    def __init__(self):
        self.buffer = BufferedByteStream()


    def computeOffset(self, start, modulus, increment):
        """
        An offset is 4 consecutive bytes encoded at C{start}.

        s = sum of bytes

        offset = (s % modulus) + increment
        """
        self.buffer.seek(start)

        offset = (
            self.buffer.read_uchar() +
            self.buffer.read_uchar() +
            self.buffer.read_uchar() +
            self.buffer.read_uchar())

        offset %= modulus
        offset += increment

        return offset


    def getDigestAndPayload(self, offset, length):
        """
        Returns the C{digest} and C{payload} components.
        """
        self.buffer.seek(0)

        payload = self.buffer.read(offset)
        digest = self.buffer.read(length)
        payload += self.buffer.read()

        return digest, payload


    def setFormat(self, format):
        self.format = format


    def setChallengeKey(self, key):
        self.challengeKey = key


    def setChallengeDigest(self, digest):
        self.challengeDigest = digest


    def decode(self, data):
        """
        Decodes the data bytes into this packet.
        """
        raise NotImplementedError

 def loads(data):
     """returns a list of messages"""
     stream = BufferedByteStream(data)
     result = list(pyamf.decode(stream=stream, encoding=3))
     stream.close()
     return result

    def write(self, previous=None):
        """
        Write (encoder) header to byte string.

        @param previous: previous header (used to compress header)
        @type previous: L{RTMPHeader}
        @return: encoded header
        @rtype: C{str}
        """
        if previous is None:
            diff = 3
        else:
            diff = self.diff(previous)

        first = self.object_id & 0x3F | ((diff ^ 3) << 6)

        if diff == 0:
            return chr(first)

        buf = BufferedByteStream()
        buf.write_uchar(first)

        buf.write_24bit_uint(self.timestamp)

        if diff > 1:
            buf.write_24bit_uint(self.length)
            buf.write_uchar(self.type)

            if diff > 2:
                buf.write_ulong(self.stream_id)

        buf.seek(0, 0)
        return buf.read()

class StreamingChannel(object):
    """
    """


    def __init__(self, channel, streamId, output):
        self.type = None
        self.channel = channel
        self.streamId = streamId
        self.output = output
        self.stream = BufferedByteStream()

        self._lastHeader = None
        self._oldStream = channel.stream
        channel.stream = self.stream

        h = header.Header(channel.channelId)

        # encode a continuation header for speed
        header.encode(self.stream, h, h)

        self._continuationHeader = self.stream.getvalue()
        self.stream.consume()


    def __del__(self):
        try:
            self.channel.stream = self._oldStream
        except:
            pass


    def setType(self, type):
        self.type = type


    def sendData(self, data, timestamp):
        c = self.channel

        if timestamp < c.timestamp:
            relTimestamp = timestamp
        else:
            relTimestamp = timestamp - c.timestamp

        h = header.Header(c.channelId, relTimestamp, self.type, len(data), self.streamId)

        if self._lastHeader is None:
            h.full = True

        c.setHeader(h)
        c.append(data)

        header.encode(self.stream, h, self._lastHeader)
        self._lastHeader = h

        c.marshallOneFrame()

        while not c.complete():
            self.stream.write(self._continuationHeader)
            c.marshallOneFrame()

        c.reset()
        self.output.write(self.stream.getvalue())
        self.stream.consume()

 def __init__(self, deferred):
     self.deferred = deferred
     self.s = BufferedByteStream()