rtpsessionmgr.java

FMJ（freedom media for java)是java视频开发的新选择

    }

    public void addTarget(SessionAddress remoteAddress)
            throws IOException {
        RTPSocketAdapter socket = new RTPSocketAdapter(
                localAddress.getDataAddress(),
                remoteAddress.getDataAddress(), remoteAddress.getDataPort(),
                remoteAddress.getTimeToLive());
        rtpHandler = new RTPHandler(this);
        rtcpHandler = new RTCPHandler(this);
        socket.getControlInputStream().setTransferHandler(rtcpHandler);
        socket.getDataInputStream().setTransferHandler(rtpHandler);
        targets.put(remoteAddress, socket);
    }

    public void initialize(RTPConnector connector) {
        try {
            rtpHandler = new RTPHandler(this);
            rtcpHandler = new RTCPHandler(this);
            connector.getControlInputStream().setTransferHandler(rtcpHandler);
            connector.getDataInputStream().setTransferHandler(rtpHandler);
            targets.put(null, connector);
            start();
        } catch (IOException e) {
            logger.log(Level.WARNING, "" + e, e);
        }
    }

    public void removeTarget(SessionAddress remoteAddress, String reason) {
        RTPConnector connector = (RTPConnector) targets.get(remoteAddress);
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        DataOutputStream output = new DataOutputStream(bytes);

        try {
            Vector sendStreams = localParticipant.getStreams();

            // Send a bye packet
            output.writeByte(0x80 | 0 | (0 & 0x1F));
            output.writeByte(RTCPPacket.PT_RR & 0xFF);
            output.writeShort(1);
            output.writeInt((int) (ssrc & 0xFFFFFFFF));
            output.writeByte(0x80 | 0 | ((sendStreams.size() + 1) & 0x1F));
            output.writeByte(RTCPPacket.PT_BYE & 0xFF);
            output.writeShort(sendStreams.size() + 1);
            output.writeInt((int) (ssrc & 0xFFFFFFFF));
            for (int i = 0; i < sendStreams.size(); i++) {
                output.writeInt((int) (((RTPSendStream)
                        sendStreams.get(i)).getSSRC() & 0xFFFFFFFF));
            }
            output.close();
            bytes.close();
            byte[] data = bytes.toByteArray();
            OutputDataStream out = connector.getControlOutputStream();
            out.write(data, 0, data.length);
        } catch (IOException e) {
            logger.log(Level.WARNING, "" + e, e);
        }
        if (connector != null) {
            targets.remove(remoteAddress);
        }
    }

    public void removeTargets(String reason) {
        Iterator iter = targets.keySet().iterator();
        while (iter.hasNext()) {
            SessionAddress addr = (SessionAddress) iter.next();
            removeTarget(addr, reason);
        }
    }

    public SendStream createSendStream(DataSource dataSource, int streamIndex)
            throws UnsupportedFormatException, IOException {
        int format = -1;
        Format fmt = null;
        double clockRate = 90000;
        if (dataSource instanceof PushBufferDataSource) {
            PushBufferStream stream =
                ((PushBufferDataSource) dataSource).getStreams()[streamIndex];
            fmt = stream.getFormat();
        } else if (dataSource instanceof PullBufferDataSource) {
            PullBufferStream stream =
                ((PullBufferDataSource) dataSource).getStreams()[streamIndex];
            fmt = stream.getFormat();
        } else {
            throw new IOException("Cannot use stream sources");
        }
        Iterator iter = formatMap.keySet().iterator();
        while (iter.hasNext()) {
            Integer id = (Integer) iter.next();
            Format testFormat = (Format) formatMap.get(id);
            if (testFormat.matches(fmt)) {
                format = id.intValue();
            }
        }
        if (format == -1) {
            throw new UnsupportedFormatException(fmt);
        }
        if (fmt instanceof AudioFormat) {
            clockRate = ((AudioFormat) fmt).getSampleRate();
        }
        Iterator iterator = targets.values().iterator();
        RTPConnector connector = (RTPConnector) iterator.next();
        OutputDataStream stream = connector.getDataOutputStream();
        return new RTPSendStream((long) (Math.random() * Integer.MAX_VALUE),
                dataSource, stream, streamIndex, localParticipant, format,
                clockRate);
    }

    public void dispose() {
        removeTargets("Quitting");
        done = true;
    }

    public Object getControl(String controlClass) {
        return null;
    }

    public Object[] getControls() {
        return new Object[] {};
    }

    /**
     * Handles an incoming RTP packet
     * @param data The packet data
     * @param offset The packet offset
     * @param length The packet length
     * @throws IOException
     */
    protected void handleRTPPacket(byte[] data, int offset, int length) {
        try {
            globalReceptionStats.addPacketRecd();
            globalReceptionStats.addBytesRecd(length);
            RTPHeader header = new RTPHeader(data, offset, length);
            long ssrc = header.getSsrc();
            Integer packetType = (Integer) ignoredStreams.get(new Long(ssrc));
            if (packetType != null) {
                if (packetType.intValue() != header.getPacketType()) {
                    ignoredStreams.remove(new Long(ssrc));
                    packetType = null;
                }
            }
            if (packetType == null) {
                RTPReceiveStream stream =
                    (RTPReceiveStream) receiveStreams.get(new Long(ssrc));
                if (stream == null) {
                    int type = header.getPacketType();
                    Format format = (Format) formatMap.get(new Integer(type));
                    if (format == null) {
                        globalReceptionStats.addUnknownType();
                        logger.warning("Unknown format identifier: "
                                + type);
                        ignoredStreams.put(new Long(ssrc), new Integer(type));
                    } else {
                        RTPDataSource dataSource =
                            new RTPDataSource(ssrc, format);
                        stream = new RTPReceiveStream(dataSource, ssrc);
                        receiveStreams.put(new Long(ssrc), stream);
                        ReceiveStreamEvent event = new NewReceiveStreamEvent(
                                this, stream);
                        new ReceiveStreamNotifier(receiveStreamListeners,
                                event);
                    }
                }
                if (stream != null) {
                    RTPDataSource dataSource =
                        (RTPDataSource) stream.getDataSource();
                    dataSource.handleRTPPacket(header,
                            data, offset + header.getSize(),
                            length - header.getSize());
                }
            }
        } catch (IOException e) {
            globalReceptionStats.addBadRTPkt();
        }
    }

    /**
     * Handles an incoming RTCP packet
     * @param data The packet data
     * @param offset The packet offset
     * @param length The packet length
     */
    protected void handleRTCPPacket(byte[] data, int offset, int length) {
        try {
            int avgeRTCPSize =
                averageRTCPSize * globalReceptionStats.getRTCPRecd();
            globalReceptionStats.addRTCPRecd();
            globalReceptionStats.addBytesRecd(length);
            averageRTCPSize =
                (avgeRTCPSize + length + 28) /
                globalReceptionStats.getRTCPRecd();
            RTCPHeader header = new RTCPHeader(data, offset, length);

            // Get the stream of the participant, if available
            long ssrc = header.getSsrc();
            RTPReceiveStream stream =
                (RTPReceiveStream) receiveStreams.get(new Long(ssrc));

            RTCPReport report = null;
            RemoteEvent remoteEvent = null;

            // If the packet is SR, read the sender info
            if (header.getPacketType() == RTCPPacket.PT_SR) {
                report = new RTCPSenderReport(data, offset, length);
                ((RTCPSenderReport) report).setStream(stream);
                remoteEvent = new SenderReportEvent(this,
                        (RTCPSenderReport) report);
                globalReceptionStats.addSRRecd();
            }

            // If the packet is RR, read the receiver info
            if (header.getPacketType() == RTCPPacket.PT_RR) {
                report = new RTCPReceiverReport(data, offset, length);
                remoteEvent = new ReceiverReportEvent(this,
                        (RTCPReceiverReport) report);
            }

            // If the report is not null
            if (report != null) {
                String cname = report.getCName();
                if (cname == null) {
                    cname = (String) senders.get(new Long(ssrc));
                }

                if (stream != null) {
                    stream.setReport(report);
                }

                // If the cname is in the report
                if (cname != null) {

                    // Store the cname for later
                    senders.put(new Long(ssrc), cname);

                    // Get the participant
                    RTPRemoteParticipant participant =
                        (RTPRemoteParticipant) activeParticipants.get(cname);
                    if (participant == null) {
                        participant = (RTPRemoteParticipant)
                                          inactiveParticipants.get(cname);
                    }

                    // If there is no participant, create one
                    if (participant == null) {
                        participant = new RTPRemoteParticipant(cname);
                        getEventLock();
                        SessionEvent event =
                            new NewParticipantEvent(this, participant);
                        new SessionNotifier(sessionListeners, event);
                        inactiveParticipants.put(cname, participant);
                    }

                    // Set the participant of the report
                    report.setParticipant(participant);
                    participant.addReport(report);

                    // If this is a bye packet, remove the stream
                    if (report.isByePacket()) {
                        participant.removeStream(stream);
                        getEventLock();
                        new ReceiveStreamNotifier(receiveStreamListeners,
                                new ByeEvent(this, participant, stream,
                                        report.getByeReason(),
                                        participant.getStreams().size() == 0));
                        if (participant.getStreams().size() == 0) {
                            activeParticipants.remove(cname);
                            inactiveParticipants.put(cname, participant);
                        }
                    } else {

                        // If the stream is not null, map the stream
                        if (stream != null) {
                            if (!activeParticipants.containsKey(cname)) {
                                inactiveParticipants.remove(cname);
                                activeParticipants.put(cname, participant);
                            }

                            if (stream.getParticipant() == null) {
                                participant.addStream(stream);
                                stream.setParticipant(participant);
                                getEventLock();
                                ReceiveStreamEvent event =
                                    new StreamMappedEvent(this, stream,
                                            participant);
                                new ReceiveStreamNotifier(
                                        receiveStreamListeners, event);
                            }
                        }
                    }
                }

                // Notify listeners of this packet
                getEventLock();
                new RemoteNotifier(remoteListeners, remoteEvent);
            } else {
                throw new IOException("Unknown report type: " +
                        header.getPacketType());
            }

        } catch (IOException e) {
            globalReceptionStats.addBadRTCPPkt();
        }
    }

    // A notifier of receive stream events
    private class ReceiveStreamNotifier extends Thread {

        // The receive stream listener
        private Vector listeners = null;

        // The event
        private ReceiveStreamEvent event = null;

        private ReceiveStreamNotifier(Vector listeners,
                ReceiveStreamEvent event) {
            this.listeners = listeners;
            this.event = event;
            start();
        }

        public void run() {
            for (int i = 0; i < listeners.size(); i++) {
                ReceiveStreamListener listener =
                    (ReceiveStreamListener) listeners.get(i);
                listener.update(event);
            }
            releaseEventLock();
        }
    }

    // A notifier of receive session events
    private class SessionNotifier extends Thread {

        // The session listener
        private Vector listeners = null;

        // The event
        private SessionEvent event = null;

        private SessionNotifier(Vector listeners,
                SessionEvent event) {
            this.listeners = listeners;
            this.event = event;
            start();
        }

        public void run() {
            for (int i = 0; i < listeners.size(); i++) {
                SessionListener listener =
                    (SessionListener) listeners.get(i);
                listener.update(event);
            }
            releaseEventLock();
        }
    }

    // A notifier of remote events
    private class RemoteNotifier extends Thread {

        // The remote listeners
        private Vector listeners = null;

        // The event
        private RemoteEvent event = null;

        private RemoteNotifier(Vector listeners,
                RemoteEvent event) {
            this.listeners = listeners;
            this.event = event;
            start();
        }

        public void run() {
            for (int i = 0; i < listeners.size(); i++) {
                RemoteListener listener =
                    (RemoteListener) listeners.get(i);
                listener.update(event);
            }
            releaseEventLock();