Merge pull request #714 from bgrozev/fixes

Fixes

Author: bgrozev

pom.xml

@@ -157,7 +157,7 @@
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>libjitsi</artifactId>
-      <version>1.0-20180829.231707-362</version>
+      <version>1.0-20180904.192703-363</version>
     </dependency>
 
     <dependency>

src/main/java/org/jitsi/videobridge/AbstractEndpoint.java

@@ -419,6 +419,7 @@ protected List<MediaStreamTrackDesc> getAllMediaStreamTracks(
         List<MediaStreamTrackDesc> allTracks = new LinkedList<>();
         channels.stream()
             .map(channel -> channel.getStream().getMediaStreamTrackReceiver())
+            .filter(Objects::nonNull)
             .forEach(
                 trackReceiver -> allTracks.addAll(
                     Arrays.asList(trackReceiver.getMediaStreamTracks())));

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

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<TransportManager> transportManagers
+                = new LinkedList<>(this.transportManagers.values());
+            transportManagers.forEach(this::closeTransportManager);
+            this.transportManagers.clear();
         }
     }
 

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());
+            }
         }
     }
 

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(

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;
     }
 
     /**