ZERO = datetime.timedelta(0)
(MARK_NUMBER, MARK_BOOL, MARK_STRING, MARK_OBJECT, MARK_MOVIECLIP, MARK_NULL,
 MARK_UNDEFINED, MARK_REFERENCE, MARK_ECMA_ARRAY, MARK_OBJECT_END,
 MARK_STRICT_ARRAY, MARK_DATE, MARK_LONG_STRING, MARK_UNSUPPORTED,
 MARK_RECORDSET, MARK_XML_DOCUMENT, MARK_TYPED_OBJECT,
 MARK_AVMPLUS_OBJECT) = range(0x12)
class UTC(datetime.tzinfo):
class OrderedDict(DictMixin, object):
class OrderedObject(OrderedDict):
class DecoderError(ValueError):
class EncoderError(ValueError):
class UndefinedType(object):
class Object(OrderedObject):
class Reference(object):
class ECMAArray(OrderedDict):
class XMLDocument(unicode):
    def utcoffset(self, dt):
    def tzname(self, dt):
    def dst(self, dt):
    def __init__(self, other=None, **kw):
    def __getitem__(self, key):
    def __setitem__(self, key, value):
    def __delitem__(self, key):
    def __iter__(self):
    def __contains__(self, key):
    def __len__(self):
        def __hash__(self):
    def keys(self):
    def iteritems(self):
    def copy(self):
    def __repr__(self):
    def __init__(self, other=None, **kw):
    def __getattr__(self, name):
    def __setattr__(self, name, value):
    def __delattr__(self, name):
    def copy(self):
    def s(self, **kw):
    def __repr__(self):
    def __eq__(self, other):
    def __ne__(self, other):
    def __hash__(self):
    def __repr__(self):
    def __init__(self, index):
    def __eq__(self, other):
    def __ne__(self, other):
    def __hash__(self):
    def __repr__(self):
def _decode_marker(s):
def _decode_number(s):
def _decode_boolean(s):
def _decode_any_string(s, unpacker):
def _decode_string(s):
def _decode_object_like(s, setter):
def _decode_object(s):
def _decode_null(s):
def _decode_undefined(s):
def _decode_reference(s):
def _decode_ecma_array(s):
def _decode_strict_array(s):
def _decode_date(s):
def _decode_long_string(s):
def _decode_xml_document(s):
def _decode_typed_object(s):
def _decode_unsupported(s):
def _decode_single(s, type_dict=decoders):
def _decode(s, type_dict=decoders):
def _encode_number(s, value):
def _encode_boolean(s, value):
def _encode_string(s, value):
def _encode_unicode(s, value):
def _encode_null(s, _value):
def _encode_undefined(s, _value):
def _encode_reference(s, value):
def _encode_strict_array(s, value):
def _encode_date(s, value):
def _encode_xml_document(s, value):
def _encode_property_name(s, value):
def _encode_object_like_content(s, value):
def _encode_ecma_array(s, value):
def _encode_object(s, value):
def _encode_single(s, value, type_dict=encoders):
def _encode(s, values, type_dict=encoders):
def _encode_variable_name(s, name):
def decode(data):
def decode_one(data):
def encode(*args):
def decode_variable(data):
def encode_variable(name, value): class Header(object):
class ChunkStreamParseError(ValueError):
class ChunkStreamValueError(ValueError):
class Demuxer(object):
class Chunker(object):
class SimpleChunkProducer(object):
class Muxer(object):
    def __init__(self, cs_id, time, size, type, ms_id, **kw):
    def __repr__(self):
        def _repr_maybe(v, provided):
def absolutize(h, base):
def read_header_head(s):
    def __init__(self, protocol):
    def doSetChunkSize(self, header, new_size):
    def doAbortMessage(self, header, cs_id):
    def doACK(self, header, seq_num):
    def doUserControlMessage(self, header, evt_type, body):
    def doWindowSize(self, header, window_size):
    def doSetBandwidth(self, header, window_size, limit_type):
    def controlMessageUnknown(self, header, body):
    def controlMessageReceived(self, header, body):
    def gen_handler(self):
def _encode_basic_header(h_type, cs_id, time=None):
def encode_full_header(cs_id, time, size, msg_type, ms_id):
def encode_comp_header(h_type, cs_id, time, size, msg_type, ms_id):
    def set_chunk_size(self, new_chunk_size):
    def __call__(self, cs_id, first_header, body, time=None):
    def __init__(self, transport):
    def queue_chunker(self, priority, chunker):
    def sync(self, priority):
    def pauseProducing(self):
    def resumeProducing(self):
    def stopProducing(self):
    def __init__(self, transport):
    def _precompute_dispatch(self, amf_ver):
    def _make_adhoc_csid(self, mt):
    def set_chunk_size(self, new_chunk_size):
    def sendMessage(self, time, type_, ms_id, body, absolute=False):
DEFAULT_CHUNK_SIZE = 128
(PROTO_SET_CHUNK_SIZE, PROTO_ABORT_MESSAGE, PROTO_ACK, PROTO_USER_CONTROL,
 PROTO_WINDOW_SIZE, PROTO_SET_BANDWIDTH, MSG_COMMAND, MSG_DATA, MSG_SO,
 MSG_AUDIO, MSG_VIDEO, MSG_AGGREGATE) = range(12) RTMP_CHUNK_SIZE = 0x01
RTMP_ABORT = 0x02
RTMP_ACK = 0x03
RTMP_USER_CONTROL = 0x04
RTMP_WINDOW_SIZE = 0x05
RTMP_SET_BANDWIDTH = 0x06
RTMP_AUDIO = 0x08
RTMP_VIDEO = 0x09
RTMP_DATA = 0x12
RTMP_SHARED_OBJ = 0x13
RTMP_COMMAND = 0x14
RTMP_AGGREGATE = 0x16
RTMP_DATA_AMF3 = 0x0f
RTMP_SHARED_OBJ_AMF3 = 0x10
RTMP_COMMAND_AMF3 = 0x11
UCTRL_STREAM_BEGIN = 0x0
UCTRL_STREAM_EOF = 0x1
UCTRL_STREAM_DRY = 0x2
UCTRL_BUFFER_LENGTH = 0x3
UCTRL_STREAM_RECORDED = 0x4
UCTRL_PING = 0x6
UCTRL_PONG = 0x7
AF_PCM_PLATFORM = 0
AF_ADPCM = 1
AF_MP3 = 2
AF_PCM_LITTLE_ENDIAN = 3
AF_NELLYMOSER_16K = 4
AF_NELLYMOSER_8K = 5
AF_NELLYMOSER = 6
AF_ALAW = 7
AF_ULAW = 8
AF_AAC = 10
AF_SPEEX = 11
AF_MP3_8K = 14
AF_DEVICE_SPECIFIC = 15
VF_JPEG = 1
VF_H263 = 2
VF_SCREEN = 3
VF_VP6 = 4
VF_VP6_ALPHA = 5
VF_SCREEN_2 = 6
VF_H264 = 7
FT_VIDEO_KEYFRAME = 1
FT_VIDEO_INTERFAME = 2
FT_VIDEO_DISPOSABLE_INTERFRAME = 3
FT_VIDEO_GENERATED_KEYFRAME = 4
FT_VIDEO_INFO = 5
FT_AAC_HEADER = 0
FT_AAC_DATA = 1
FT_H264_HEADER = 0
FT_H264_NALU = 1
FT_H264_EOS = 2 class Demuxer(object):
    chunk_size = 128

    def __init__(self, protocol):
        self.chstr_map = {}
        self.msg_map = {}
        self.protocol = protocol

        # handlers: {type => (verify_size, cnv_size, cnv_func, pass_rest,
        #                     handler_func)}
        # TODO: should some of those be handled directly by a higer layer?...
        self.ctrl_handlers = {
            0x1: (_s_ulong_b.size, _s_ulong_b.size, _s_ulong_b.unpack, False,
                  self.doSetChunkSize),
            0x2: (_s_ulong_b.size, _s_ulong_b.size, _s_ulong_b.unpack, False,
                  self.doAbortMessage),
            0x3: (_s_ulong_b.size, _s_ulong_b.size, _s_ulong_b.unpack, False,
                  self.doACK),
            0x4: (None,            _s_ushort.size,  _s_ushort.unpack,  True,
                  self.doUserControlMessage),
            0x5: (_s_ulong_b.size, _s_ulong_b.size, _s_ulong_b.unpack, False,
                  self.doWindowSize),
            0x6: (_s_set_bw.size,  _s_set_bw.size,  _s_set_bw.unpack,  False,
                  self.doSetBandwidth),
            }

    def doSetChunkSize(self, header, new_size):
        if not new_size > 0:
            raise ChunkStreamValueError('set chunk size: '
                                        'need positive chunk size')
        self.chunk_size = new_size

    def doAbortMessage(self, header, cs_id):
        chunks_acc = self.chstr_map.pop(cs_id, None)
        if chunks_acc:
            self.msg_map[cs_id] = chunks_acc[0]

    def doACK(self, header, seq_num):
        pass

    def doUserControlMessage(self, header, evt_type, body):
        pass

    def doWindowSize(self, header, window_size):
        pass

    def doSetBandwidth(self, header, window_size, limit_type):
        pass

    def controlMessageUnknown(self, header, body):
        # this should only be invoked on undocumented control message 7
        raise NotImplementedError('TODO!')

    def controlMessageReceived(self, header, body):
        handler = self.ctrl_handlers.get(header.type)
        if not handler:
            self.controlMessageUnknown(header, body)
            return

        verify_size, cnv_size, cnv_func, pass_rest, handler_func = handler
        if verify_size is not None:
            if len(body) != verify_size:
                raise ChunkStreamParseError(('expected msg of size: %d, '
                                             'got: %d') % (verify_size,
                                                           len(body)))

        args = (header,) + cnv_func(body.read(cnv_size))
        if pass_rest:
            args += (body,)

        handler_func(*args)

    def gen_handler(self):
        s = ''

        while 1:
            if len(s) < 1:
                s = yield 1     # need 1 byte to read "basic header"

            htype, csid = read_header_head(s)

            csid_sel = min(csid, 2)
            head_size = _sizes_1[csid_sel] + _sizes_2[htype]

            if len(s) < head_size:
                s = yield head_size # bytes to read the rest of the header

            if csid == 0:
                csid = _s_uchar.unpack(s.read(1))[0] + 64
            elif csid == 1:
                csid = _s_ushort_l.unpack(s.read(2))[0] + 64

            m_time, m_size, m_type, m_msid = None, None, None, None
            c_h, h_base, accbody, to_read = None, None, None, None
            m_time_ext = False
            h_absolute = (htype == 0)

            # check csid in the message and chunk stream caches
            if csid in self.chstr_map:
                c_h, accbody, to_read = self.chstr_map[csid]
                m_time_ext = (c_h.time >= 0x00ffffff)
            elif not h_absolute and csid in self.msg_map:
                h_base = self.msg_map[csid]
                m_time_ext = (h_base.time >= 0x00ffffff)
            else:
                # TODO: check/warn header is absolute here
                pass

            if htype == 3:
                pass
            elif htype == 2:
                _time_1, m_time = _s_time.unpack(s.read(3))
                m_time += _time_1 << 8
            else:
                (_time_1, m_time, _size_1,
                 m_size, m_type) = _s_time_size_type.unpack(s.read(7))
                m_time += _time_1 << 8
                m_size += _size_1 << 8
                if htype == 0:
                    m_msid, = _s_ulong_l.unpack(s.read(4))

            if m_time == 0x00ffffff or m_time is None and m_time_ext:
                if len(s) < 4:
                    s = yield 4 # 4 bytes of "extended timestamp"
                m_time, = _s_ulong_b.unpack(s.read(4))

            h = Header(csid, m_time, m_size, m_type, m_msid)

            # fill header if cached entry was found earlier
            if c_h:
                # TODO: check/warn header consistency (with the cached
                # first one)
                h = c_h
            elif h_base:
                h = absolutize(h, h_base)
                to_read = h.size
            else:
                to_read = h.size

            need_bytes = min(to_read, self.chunk_size)
            if need_bytes > 0:
                if len(s) < need_bytes:
                    s = yield need_bytes

                if accbody is None:
                    accbody = s.read_seq(need_bytes)
                else:
                    accbody += s.read_seq(need_bytes)
            else:
                if accbody is None:
                    accbody = []

            if to_read > self.chunk_size:
                self.chstr_map[h.cs_id] = (h, accbody,
                                           to_read - self.chunk_size)
            else:
                self.chstr_map.pop(h.cs_id, None)
                self.msg_map[h.cs_id] = h

                if h.cs_id == 2 and 0 < h.type < 8 and h.ms_id == 0:
                    self.controlMessageReceived(h, vecbuf.VecBuf(accbody))
                else:
                    self.protocol.messageReceived(h, vecbuf.VecBuf(accbody)) class Muxer(object):
    chunker_class = Chunker
    chunk_producer_class = SimpleChunkProducer

    AMF_ver = AMF_v0

    def __init__(self, transport):
        # (dynamic) explicitly reserved chunk stream ids:
        # { (ms_id, msg_type) => cs_id }
        self._reserved_csids = {}

        # dynamically allocated, purgeable, chunk stream id cache:
        # { (ms_id, msg_type) => cs_id }
        self._adhoc_csids = {}
        # note: purging needs to be done carefully, since reusing
        # cs_id for a chunk stream with a higher priority might
        # corrupt chunk order if our "consumer" has buffered chunks
        # and is actually prioritizing the chunks

        # used/sent header cache:
        # { cs_id => (abs_time, time_diff, size, msg_type, ms_id) }
        self._cached = {}

        self._chunker = self.chunker_class()
        self.producer = self.chunk_producer_class(transport)

        self._precompute_dispatch(self.AMF_ver)

    def _precompute_dispatch(self, amf_ver):
        self._type_dispatch = [elt[amf_ver] for elt in msg_type_dispatch]

    def _make_adhoc_csid(self, mt):
        # this is a simple, not an optimal implementation...
        cs_id_r = 2
        cs_id_t = 2

        if self._reserved_csids:
            cs_id_r = max(self._reserved_csids.values())
        if self._adhoc_csids:
            cs_id_t = max(self._adhoc_csids.values())

        cs_id = max(cs_id_r, cs_id_t) + 1
        self._adhoc_csids[mt] = cs_id
        return cs_id

    def set_chunk_size(self, new_chunk_size):
        """Should be called immediately after sending/queueing
        PROTO_SET_CHUNK_SIZE message.
        """
        # force complete chunking of any messages queued in the
        # producer with priority at least equal to priority of
        # PROTO_SET_CHUNK_SIZE (0)
        self.producer.sync(0)

        # now it should be safe to change the chunk size
        self._chunker.set_chunk_size(new_chunk_size)

    def sendMessage(self, time, type_, ms_id, body, absolute=False):
        """Build and send binary representation of message.

        @param time: absolute time of the message, in milliseconds, in
                     [0, 0xffffffff] range
        @type time: int

        @param type_: type of message, one from PROTO_* and MSG_* constants

        @param ms_id: message stream id, should be 0 for PROTO_* messages

        @param body: the payload of the message
        @type body:  VecBuf
        """
        size = len(body)

        # first: priority based on (abstracted) message type
        priority = 0x10
        if type_ == MSG_VIDEO:
            priority += 0x10

        # second: get the actual message type, depending on the
        # negotiated AMF version
        msg_type = self._type_dispatch[type_]

        # third, figure out cs_id
        cs_id = None
        if msg_type < 0x08:
            # protocol message: use reserved chunk stream 2
            cs_id = 2
            # ... make sure the stream id is 0
            ms_id = 0
            # ... increase the priority
            priority -= 0x10
            # ... and don't compress the header at all
            absolute = True
        else:
            mt = (ms_id, msg_type)
            # note, valid chunk stream id cannot be 0
            cs_id = (self._reserved_csids.get(mt) or
                     self._adhoc_csids.get(mt) or
                     self._make_adhoc_csid(mt))

        if absolute or cs_id not in self._cached:
            self._cached[cs_id] = (time, time, size, msg_type, ms_id)
            raw_header = encode_full_header(cs_id, time, size, msg_type, ms_id)
        else:
            (h_abs_time, h_time, h_size, h_msg_type,
             h_ms_id) = self._cached[cs_id]

            if ms_id == h_ms_id:
                # header is relative, compress time
                new_time = time - h_abs_time # use just the diff

                if new_time < 0:
                    # could happen after a seek...? :/ - don't compress
                    new_time = time
                    h_type = 0
                else:
                    if msg_type == h_msg_type and size == h_size:
                        if new_time == h_time:
                            h_type = 3
                        else:
                            h_type = 2
                    else:
                        h_type = 1
            else:
                # absolute header here, use absolute time
                new_time = time
                h_type = 0

            self._cached[cs_id] = (time, new_time, size, msg_type, ms_id)
            raw_header = encode_comp_header(h_type, cs_id, new_time, size,
                                            msg_type, ms_id)
            time = new_time

        self.producer.queue_chunker(priority,
                                    self._chunker(cs_id, raw_header, body,
                                                  time)) class ProtocolContractError(ValueError):
    pass class Handshaker(object):
    """The handshake protocol.

    Interaction should go something like this:
      * cl: ()->v+CLHS sent->(v+SVHS rcvd)->SVHS sent->(CLHS rcvd) OK!
      * sv: ()->(v rcvd)->v+SVHS sent->(CLHS rcvd)->CLHS sent->(SVHS rcvd) OK!

    where:
      * v    - version packet
      * SVHS - server handshake packet
      * CLHS - client handshake packet
    """

    packet_bytes = 1536
    protocol_version = 3

    def __init__(self, protocol, epoch_base, is_client=False):
        self.epoch = epoch_base
        self.version_packet = _s_uchar.pack(self.protocol_version)
        self.protocol = protocol
        self.is_client = is_client

    def epoch_time(self):
        return time.time() - self.epoch


    def defaultHandshakeFailed(self, msg=''):
        self.protocol.handshakeFailed()
        # raising exception should cause closing the connection eventually
        raise HandshakeFailedError(string=msg)


    def version_supported(self, version):
        return version == self.protocol_version

    def generate_request(self, context=None):
        return (_s_ts_simpver.pack(ms_time(self.epoch_time()), 0) +
                generate_random_bytes(self.packet_bytes - 8))

    def generate_response(self, request):
        return (buffer(request, 0, 4) +
                (buffer(_s_ulong.pack(ms_time(self.epoch_time()))) +
                 buffer(request, 8)))

    def verify_response(self, request, response):
        return (buffer(request, 0, 4) == buffer(response, 0, 4) and
                buffer(request, 8) == buffer(response, 8))

    def gen_handler(self):
        p = self.protocol

        if self.is_client:
            ts_init = time.time()
            hs_mine = self.generate_request()
            p.transport.writeSequence((self.version_packet, hs_mine))

        # wait for the "version packet"
        s = yield 1
        version_other, = _s_uchar.unpack(s.read(1))

        if not self.version_supported(version_other):
            self.defaultHandshakeFailed('protocol version mismatch')
            return

        # now, the primary handshake packet
        s = yield self.packet_bytes

        hs_other = s.read(self.packet_bytes)

        if not self.is_client:
            ts_init = time.time()
            hs_mine = self.generate_request(context=hs_other)
            p.transport.writeSequence((self.version_packet, hs_mine))

        p.transport.write(self.generate_response(hs_other))

        # and now, the handshake response
        s = yield self.packet_bytes

        hs_response = buffer(s.read(self.packet_bytes))
        ts_received_back = time.time()

        if not self.verify_response(hs_mine, hs_response):
            self.defaultHandshakeFailed('response verification failed')
            return

        p.handshakeSucceeded(ts_init - self.epoch, ts_received_back - ts_init) class GeneratorWrapperProtocol(protocol.Protocol):
    def __init__(self, proto=None):
        self._buf = vecbuf.VecBuf()
        self._toread = 0

        self._handler_changed = False

        self._handler = proto
        if proto:
            self._toread = self._handler.send(None)

    def init_handler(self, proto=None, do_init=True):
        if proto:
            if self._handler:
                # need to indicate we should retry on StopIteration...
                self._handler_changed = True
            self._handler = proto
        if do_init:
            self._toread = self._handler.send(None)

    def connectionLost(self, reason=protocol.connectionDone):
        # clean up
        self._buf.read(len(self._buf))
        self._handler = None
        self._toread = None

    def dataReceived(self, data):
        self._buf.write(data)

        while 1: # need to be able to retry immediately when handlers change
            try:
                while len(self._buf) >= self._toread:
                    self._toread = self._handler.send(self._buf)
                break           # no data, no retry
            except StopIteration:
                if self._handler_changed:
                    self._handler_changed = False
                else:
                    self._toread = None
                    self._handler = None
                    break
            except:
                # undocumented, but real; this way we can signal
                # protocol errors directly, and by not passing/raising
                # directly we avoid a stack trace log, with the same
                # effect - connectionLost
                f = failure.Failure()
                log.info(f.value, exc_info=log.isEnabledFor(logging.DEBUG))
                return f def semiflatten(bufvec):
    """Return a copy of the given sequence, with all elements rendered
    to strings. Meant to be used with a sequence of buffers.

    @type bufvec: sequence of buffer

    @rtype: sequence of str
    """
    return [elt[:] for elt in bufvec] def vb(s):
    # TODO: add this functionality to the vecbuf/VecBuf itself
    return VecBuf([s])
import struct
import twimp.log
from twisted.internet import protocol
from twisted.internet.protocol import Factory
from twimp import amf0
from twimp import chunks
from twimp import const
from twimp.chunks import Demuxer, Muxer
from twimp.error import ProtocolContractError
from twimp.handshake import Handshaker
from twimp.primitives import _s_ulong_b as _s_ulong, _s_double_ulong_b
from twimp.utils import GeneratorWrapperProtocol
from twimp.vecbuf import semiflatten
from twimp.helpers import vb
class UserControlDispatchDemuxer(Demuxer):
    def __init__(self, protocol):
        super(UserControlDispatchDemuxer, self).__init__(protocol)

        # { type => (verify_size, cnv_size, cnv_func, handler_func) }
        self.user_ctrl_handlers = None
        self.build_user_control_dispatch()

    def build_user_control_dispatch(self):
        self.user_ctrl_handlers = {
            const.UCTRL_STREAM_BEGIN:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlStreamBegin),
            const.UCTRL_STREAM_EOF:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlStreamEOF),
            const.UCTRL_STREAM_DRY:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlStreamDry),
            const.UCTRL_BUFFER_LENGTH:
                (_s_double_ulong_b.size, _s_double_ulong_b.size,
                 _s_double_ulong_b.unpack, self.doUserControlBufferLength),
            const.UCTRL_STREAM_RECORDED:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlStreamRecorded),
            const.UCTRL_PING:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlPing),
            const.UCTRL_PONG:
                (_s_ulong.size, _s_ulong.size, _s_ulong.unpack,
                 self.doUserControlPong),
            }

    def doUserControlMessage(self, header, evt_type, body):
        handler = self.user_ctrl_handlers.get(evt_type)
        if not handler:
            self.doUserControlUnknownType(header, evt_type, body)
            return

        verify_size, cnv_size, cnv_func, handler_func = handler
        if len(body) != verify_size:
            raise ProtocolContractError(('expected user ctrl msg of size: %d, '
                                         'got: %d') % (verify_size,
                                                       len(body)))

        args = (header,) + cnv_func(body.read(cnv_size))
        handler_func(*args)

    # handling pings automatically
    def doUserControlPing(self, header, peer_time):
        sm = self.protocol.muxer.sendMessage
        sm(0, chunks.PROTO_USER_CONTROL, 0,
           vb(_s_uctrl_single.pack(const.UCTRL_PONG, peer_time)))

    # ... implementation of the rest left to the user
    def doUserControlStreamBegin(self, header, stream_id):
        pass

    def doUserControlStreamEOF(self, header, stream_id):
        pass

    def doUserControlStreamDry(self, header, stream_id):
        pass

    def doUserControlStreamRecorded(self, header, stream_id):
        pass

    def doUserControlBufferLength(self, header, stream_id, length):
        pass

    def doUserControlPong(self, header, echo_time):
        pass

    def doUserControlUnknownType(self, header, evt_type, body):
        pass


class DispatchProtocol(BaseProtocol):
    demuxer_class = UserControlDispatchDemuxer

    def __init__(self):
        BaseProtocol.__init__(self)

        self.msg_dispatch = None

        self.bytes_read = 0 # all bytes received minus headers (?)
        self._next_ack = 0
        self.window_size = 2500000

        self.set_next_ack()

        self.build_message_dispatch()

    def set_window_size(self, new_size):
        if new_size != self.window_size:
            old_size, self.window_size = self.window_size, new_size
            self.set_next_ack(old_size)
            self.check_send_ack()

    def set_next_ack(self, old_window_size=0):
        # the actual "algorithm" is here - Adobe's software seems to send
        # acks after _half_ the window size...
        old_ack_inc = int(old_window_size / 2)
        ack_inc = int(self.window_size / 2)
        self._next_ack += ack_inc - old_ack_inc

    def check_send_ack(self):
        if self._next_ack < self.bytes_read:
            self.set_next_ack()
            self.muxer.sendMessage(0, chunks.PROTO_ACK, 0,
                                   vb(_s_ulong.pack(self.bytes_read)))

    def bytes_received(self, count):
        self.bytes_read += count
        self.check_send_ack()

    def build_message_dispatch(self):
        # for now we only support AMF0

        def amf_args(header, body):

            return header.abs_time, header.ms_id, amf0.decode(body)