diff --git a/pom.xml b/pom.xml index 3ad58cba57..525276b2ff 100644 --- a/pom.xml +++ b/pom.xml @@ -157,7 +157,7 @@ ${project.groupId} libjitsi - 1.0-20180829.231707-362 + 1.0-20180904.192703-363 diff --git a/src/main/java/org/jitsi/videobridge/AbstractEndpoint.java b/src/main/java/org/jitsi/videobridge/AbstractEndpoint.java index edeb973fa3..2d702e79d3 100644 --- a/src/main/java/org/jitsi/videobridge/AbstractEndpoint.java +++ b/src/main/java/org/jitsi/videobridge/AbstractEndpoint.java @@ -419,6 +419,7 @@ protected List getAllMediaStreamTracks( List allTracks = new LinkedList<>(); channels.stream() .map(channel -> channel.getStream().getMediaStreamTrackReceiver()) + .filter(Objects::nonNull) .forEach( trackReceiver -> allTracks.addAll( Arrays.asList(trackReceiver.getMediaStreamTracks()))); diff --git a/src/main/java/org/jitsi/videobridge/AbstractEndpointMessageTransport.java b/src/main/java/org/jitsi/videobridge/AbstractEndpointMessageTransport.java index 2b534b3274..79b9269935 100644 --- a/src/main/java/org/jitsi/videobridge/AbstractEndpointMessageTransport.java +++ b/src/main/java/org/jitsi/videobridge/AbstractEndpointMessageTransport.java @@ -38,7 +38,7 @@ public abstract class AbstractEndpointMessageTransport * information. */ private static final Logger classLogger - = Logger.getLogger(EndpointMessageTransport.class); + = Logger.getLogger(AbstractEndpointMessageTransport.class); /** * The {@link Endpoint} associated with this diff --git a/src/main/java/org/jitsi/videobridge/Conference.java b/src/main/java/org/jitsi/videobridge/Conference.java index 3425e19bfe..7502734e69 100644 --- a/src/main/java/org/jitsi/videobridge/Conference.java +++ b/src/main/java/org/jitsi/videobridge/Conference.java @@ -491,8 +491,10 @@ private void closeTransportManagers() { synchronized (transportManagers) { - transportManagers.forEach((id, tm) -> closeTransportManager(tm)); - transportManagers.clear(); + Collection transportManagers + = new LinkedList<>(this.transportManagers.values()); + transportManagers.forEach(this::closeTransportManager); + this.transportManagers.clear(); } } diff --git a/src/main/java/org/jitsi/videobridge/RtpChannel.java b/src/main/java/org/jitsi/videobridge/RtpChannel.java index 0c8011d336..58f8b5403c 100644 --- a/src/main/java/org/jitsi/videobridge/RtpChannel.java +++ b/src/main/java/org/jitsi/videobridge/RtpChannel.java @@ -1903,19 +1903,29 @@ private void updateStatisticsOnExpire() updatePacketsAndBytes(conferenceStatistics); - TrackStats streamStats - = stream.getMediaStreamStats().getSendStats(); - logger.info(Logger.Category.STATISTICS, - "expire_ch_stats," + getLoggingId() + - " bRecv=" + statistics.bytesReceived + - ",bSent=" + statistics.bytesSent + - ",pRecv=" + statistics.packetsReceived + - ",pSent=" + statistics.packetsSent + - ",bRetr=" + streamStats.getBytesRetransmitted() + - ",bNotRetr=" + streamStats.getBytesNotRetransmitted() + - ",pRetr=" + streamStats.getPacketsRetransmitted() + - ",pNotRetr=" + streamStats.getPacketsNotRetransmitted() + - ",pMiss=" + streamStats.getPacketsMissingFromCache()); + MediaStream stream = this.stream; + TrackStats streamStats; + if (stream != null && + (streamStats = stream.getMediaStreamStats().getSendStats()) + != null) + { + logger.info(Logger.Category.STATISTICS, + "expire_ch_stats," + getLoggingId() + + " bRecv=" + statistics.bytesReceived + + ",bSent=" + statistics.bytesSent + + ",pRecv=" + statistics.packetsReceived + + ",pSent=" + statistics.packetsSent + + ",bRetr=" + + streamStats.getBytesRetransmitted() + + ",bNotRetr=" + + streamStats.getBytesNotRetransmitted() + + ",pRetr=" + + streamStats.getPacketsRetransmitted() + + ",pNotRetr=" + + streamStats.getPacketsNotRetransmitted() + + ",pMiss=" + + streamStats.getPacketsMissingFromCache()); + } } } diff --git a/src/main/java/org/jitsi/videobridge/cc/SimulcastController.java b/src/main/java/org/jitsi/videobridge/cc/SimulcastController.java index da617dd39b..091aea3583 100644 --- a/src/main/java/org/jitsi/videobridge/cc/SimulcastController.java +++ b/src/main/java/org/jitsi/videobridge/cc/SimulcastController.java @@ -1525,10 +1525,27 @@ boolean accept(FrameDesc source, RawPacket pkt) private RawPacket handleVp8PictureIdRewriting(RawPacket pktOut) { MediaStreamTrackDesc source = weakSource.get(); - assert source != null; + if (source == null) + { + logger.error("Source is null, dropping packet."); + return null; + } - REDBlock redBlock = source.getMediaStreamTrackReceiver() - .getStream().getPrimaryREDBlock(pktOut); + MediaStreamTrackReceiver trackReceiver + = source.getMediaStreamTrackReceiver(); + if (trackReceiver == null) + { + logger.error("Track receiver is null, dropping packet."); + return null; + } + + MediaStream stream = trackReceiver.getStream(); + if (stream == null) + { + logger.error("Stream is null, dropping packet."); + return null; + } + REDBlock redBlock = stream.getPrimaryREDBlock(pktOut); if (!DePacketizer .VP8PayloadDescriptor.hasExtendedPictureId( diff --git a/src/main/java/org/jitsi/videobridge/octo/OctoTransportManager.java b/src/main/java/org/jitsi/videobridge/octo/OctoTransportManager.java index 333b8c5ae6..376f203ca4 100644 --- a/src/main/java/org/jitsi/videobridge/octo/OctoTransportManager.java +++ b/src/main/java/org/jitsi/videobridge/octo/OctoTransportManager.java @@ -51,6 +51,11 @@ public class OctoTransportManager */ public static final String NAMESPACE = "http://jitsi.org/octo"; + /** + * The timeout to set on the sockets that we create. + */ + private static final int SO_TIMEOUT = 1000; + /** * Converts a "relay ID" to a socket address. The current implementation * assumes that the ID has the form of "address:port". @@ -239,7 +244,7 @@ private void initializeSockets() private DatagramSocket createOctoSocket(DatagramSocket socket) throws SocketException { - return new DelegatingDatagramSocket(socket) + DatagramSocket s = new DelegatingDatagramSocket(socket) { @Override public void receive(DatagramPacket p) @@ -274,6 +279,18 @@ public void send(DatagramPacket p) doSend(p, true); } }; + + // With the hierarchy of sockets that we use for Octo (Delegating -> + // Multiplexed -> Multiplexing -> DatagramSocket) the calls receive() + // are handled by the Multiplexing instance. Since it is persistent, it + // will not get closed when this socket instance is closed, and will + // therefore not throw a SocketClosedException. This means that we can + // not rely on this exception to stop the receive thread + // (RTPConnectorInputStream#receiveThread), and therefore we need a + // finite timeout. + s.setSoTimeout(SO_TIMEOUT); + + return s; } /**