/* * Copyright @ 2015 - Present, 8x8 Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package org.jitsi.videobridge; import edu.umd.cs.findbugs.annotations.*; import org.jetbrains.annotations.*; import org.jetbrains.annotations.Nullable; import org.jitsi.eventadmin.*; import org.jitsi.nlj.*; import org.jitsi.rtp.*; import org.jitsi.rtp.rtcp.rtcpfb.payload_specific_fb.*; import org.jitsi.rtp.rtp.*; import org.jitsi.utils.collections.*; import org.jitsi.utils.event.*; import org.jitsi.utils.logging.DiagnosticContext; import org.jitsi.utils.logging2.*; import org.jitsi.utils.logging2.Logger; import org.jitsi.utils.logging2.LoggerImpl; import org.jitsi.videobridge.octo.*; import org.jitsi.videobridge.shim.*; import org.jitsi.videobridge.util.*; import org.jitsi.xmpp.extensions.colibri.*; import org.json.simple.*; import org.jxmpp.jid.*; import org.jxmpp.jid.parts.*; import org.osgi.framework.*; import java.beans.*; import java.io.*; import java.lang.*; import java.lang.SuppressWarnings; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; import java.util.logging.*; import static org.jitsi.utils.collections.JMap.entry; import static org.jitsi.videobridge.EndpointMessageBuilder.*; /** * Represents a conference in the terms of Jitsi Videobridge. * * @author Lyubomir Marinov * @author Boris Grozev * @author Hristo Terezov * @author George Politis */ public class Conference extends PropertyChangeNotifier implements PropertyChangeListener, Expireable, AbstractEndpointMessageTransport.EndpointMessageTransportEventHandler { /** * The endpoints participating in this {@link Conference}. Although it's a * {@link ConcurrentHashMap}, writing to it must be protected by * synchronizing on the map itself, because it must be kept in sync with * {@link #endpointsCache}. */ private final ConcurrentHashMap endpoints = new ConcurrentHashMap<>(); /** * A read-only cache of the endpoints in this conference. Note that it * contains only the {@link Endpoint} instances (and not Octo endpoints). * This is because the cache was introduced for performance reasons only * (we iterate over it for each RTP packet) and the Octo endpoints are not * needed. */ private List endpointsCache = Collections.emptyList(); private final Object endpointsCacheLock = new Object(); /** * The {@link EventAdmin} instance (to be) used by this {@code Conference} * and all instances (of {@code Content}, {@code Channel}, etc.) created by * it. */ private final EventAdmin eventAdmin; /** * The indicator which determines whether {@link #expire()} has been called * on this Conference. */ @SuppressFBWarnings( value = "IS2_INCONSISTENT_SYNC", justification = "The value is deemed safe to read without " + "synchronization.") private boolean expired = false; /** * The JID of the conference focus who has initialized this instance and * from whom requests to manage this instance must come or they will be * ignored. If null value is assigned we don't care who modifies * the conference. */ private final Jid focus; /** * The (unique) identifier/ID of this instance. */ private final String id; /** * The "global" id of this conference, set by the controller (e.g. jicofo) * as opposed to the bridge. This defaults to {@code null} unless it is * specified. */ private final String gid; /** * The world readable name of this instance if any. */ private Localpart name; /** * The time in milliseconds of the last activity related to this * Conference. In the time interval between the last activity and * now, this Conference is considered inactive. */ @SuppressFBWarnings( value = "IS2_INCONSISTENT_SYNC", justification = "The value is deemed safe to read without " + "synchronization.") private long lastActivityTime; /** * If {@link #focus} is null the value of the last known focus is * stored in this member. */ private Jid lastKnownFocus; /** * The PropertyChangeListener which listens to * PropertyChangeEvents on behalf of this instance while * referencing it by a WeakReference. */ private final PropertyChangeListener propertyChangeListener = new WeakReferencePropertyChangeListener(this); /** * The speech activity (representation) of the Endpoints of this * Conference. */ private final ConferenceSpeechActivity speechActivity; /** * The audio level listener. */ private final AudioLevelListener audioLevelListener; /** * The Videobridge which has initialized this Conference. */ private final Videobridge videobridge; /** * Holds conference statistics. */ private final Statistics statistics = new Statistics(); /** * The {@link Logger} to be used by this instance to print debug * information. */ private final Logger logger; /** * Whether this conference should be considered when generating statistics. */ private final boolean includeInStatistics; /** * The time when this {@link Conference} was created. */ private final long creationTime = System.currentTimeMillis(); /** * The {@link ExpireableImpl} which we use to safely expire this conference. */ private final ExpireableImpl expireableImpl; /** * The shim which handles Colibri-related logic for this conference. */ private final ConferenceShim shim; //TODO not public final public EncodingsManager encodingsManager = new EncodingsManager(); /** * This {@link Conference}'s link to Octo. */ private ConfOctoTransport tentacle; /** * Initializes a new Conference instance which is to represent a * conference in the terms of Jitsi Videobridge which has a specific * (unique) ID and is managed by a conference focus with a specific JID. * * @param videobridge the Videobridge on which the new * Conference instance is to be initialized * @param id the (unique) ID of the new instance to be initialized * @param focus the JID of the conference focus who has requested the * initialization of the new instance and from whom further/future requests * to manage the new instance must come or they will be ignored. * Pass null to override this safety check. * @param name world readable name of this instance if any. * @param enableLogging whether logging should be enabled for this * {@link Conference} and its sub-components, and whether this conference * should be considered when generating statistics. * @param gid the optional "global" id of the conference. */ public Conference(Videobridge videobridge, String id, Jid focus, Localpart name, boolean enableLogging, String gid) { this.videobridge = Objects.requireNonNull(videobridge, "videobridge"); Level minLevel = enableLogging ? Level.ALL : Level.WARNING; Map context = JMap.ofEntries( entry("confId", id) ); if (gid != null) { context.put("gid", gid); } if (name != null) { context.put("conf_name", name.toString()); } logger = new LoggerImpl(Conference.class.getName(), minLevel, new LogContext(context)); this.shim = new ConferenceShim(this, logger); this.id = Objects.requireNonNull(id, "id"); this.gid = gid; this.focus = focus; this.eventAdmin = enableLogging ? videobridge.getEventAdmin() : null; this.includeInStatistics = enableLogging; this.name = name; lastKnownFocus = focus; speechActivity = new ConferenceSpeechActivity(this); audioLevelListener = (sourceSsrc, level) -> speechActivity.levelChanged(sourceSsrc, (int) level); expireableImpl = new ExpireableImpl(logger, this::expire); if (enableLogging) { eventAdmin.sendEvent(EventFactory.conferenceCreated(this)); Videobridge.Statistics videobridgeStatistics = videobridge.getStatistics(); videobridgeStatistics.totalConferencesCreated.incrementAndGet(); } // We listen to our own events so we have a centralized place to handle // certain things (e.g. anytime the endpoints list changes) addPropertyChangeListener(propertyChangeListener); touch(); } /** * Creates a new diagnostic context instance that includes the conference * name and the conference creation time. * * @return the new {@link DiagnosticContext} instance. */ public DiagnosticContext newDiagnosticContext() { if (name != null) { DiagnosticContext diagnosticContext = new DiagnosticContext(); diagnosticContext.put("conf_name", name.toString()); diagnosticContext.put("conf_creation_time_ms", creationTime); return diagnosticContext; } else { return new NoOpDiagnosticContext(); } } /** * Gets the statistics of this {@link Conference}. * * @return the statistics of this {@link Conference}. */ public Statistics getStatistics() { return statistics; } /** * @return whether this conference should be included in generated * statistics. */ public boolean includeInStatistics() { return includeInStatistics; } /** * Sends a message to a subset of endpoints in the call, primary use * case being a message that has originated from an endpoint (as opposed to * a message originating from the bridge and being sent to all endpoints in * the call, for that see {@link #broadcastMessage(String)}. * * @param msg the message to be sent * @param endpoints the list of Endpoints to which the message will * be sent. */ public void sendMessage( String msg, List endpoints, boolean sendToOcto) { for (AbstractEndpoint endpoint : endpoints) { try { endpoint.sendMessage(msg); } catch (IOException e) { logger.error( "Failed to send message on data channel to: " + endpoint.getID() + ", msg: " + msg, e); } } if (sendToOcto && tentacle != null) { tentacle.sendMessage(msg); } } /** * Used to send a message to a subset of endpoints in the call, primary use * case being a message that has originated from an endpoint (as opposed to * a message originating from the bridge and being sent to all endpoints in * the call, for that see {@link #broadcastMessage(String)}. * * @param msg the message to be sent * @param endpoints the list of Endpoints to which the message will * be sent. */ public void sendMessage(String msg, List endpoints) { sendMessage(msg, endpoints, false); } /** * Broadcasts a string message to all endpoints of the conference. * * @param msg the message to be broadcast. */ public void broadcastMessage(String msg, boolean sendToOcto) { sendMessage(msg, getEndpoints(), sendToOcto); } /** * Broadcasts a string message to all endpoints of the conference. * * @param msg the message to be broadcast. */ public void broadcastMessage(String msg) { broadcastMessage(msg, false); } /** * Requests a keyframe from the endpoint with the specified id, if the * endpoint is found in the conference. * * @param endpointID the id of the endpoint to request a keyframe from. */ public void requestKeyframe(String endpointID, long mediaSsrc) { AbstractEndpoint remoteEndpoint = getEndpoint(endpointID); if (remoteEndpoint != null) { remoteEndpoint.requestKeyframe(mediaSsrc); } else if (logger.isDebugEnabled()) { logger.debug( "Cannot request keyframe because the endpoint was not found."); } } /** * Sets the values of the properties of a specific * ColibriConferenceIQ to the values of the respective * properties of this instance. Thus, the specified iq may be * thought of as a description of this instance. *

* Note: The copying of the values is shallow i.e. the * Contents of this instance are not described in the specified * iq. *

* * @param iq the ColibriConferenceIQ to set the values of the * properties of this instance on */ public void describeShallow(ColibriConferenceIQ iq) { iq.setID(getID()); iq.setName(getName()); } /** * Notifies this instance that {@link #speechActivity} has identified a * speaker switch event in this multipoint conference and there is now a new * dominant speaker. */ void dominantSpeakerChanged() { AbstractEndpoint dominantSpeaker = speechActivity.getDominantEndpoint(); if (logger.isInfoEnabled()) { String id = dominantSpeaker == null ? "null" : dominantSpeaker.getID(); logger.info("ds_change ds_id=" + id); getVideobridge().getStatistics().totalDominantSpeakerChanges.increment(); } speechActivityEndpointsChanged(speechActivity.getEndpointIds()); if (dominantSpeaker != null) { broadcastMessage( createDominantSpeakerEndpointChangeEvent( dominantSpeaker.getID())); if (getEndpointCount() > 2) { double senderRtt = getRtt(dominantSpeaker); double maxReceiveRtt = getMaxReceiverRtt(dominantSpeaker.getID()); // We add an additional 10ms delay to reduce the risk of the keyframe arriving // too early double keyframeDelay = maxReceiveRtt - senderRtt + 10; if (logger.isDebugEnabled()) { logger.debug("Scheduling keyframe request from " + dominantSpeaker.getID() + " after a delay" + " of " + keyframeDelay + "ms"); } TaskPools.SCHEDULED_POOL.schedule( (Runnable)dominantSpeaker::requestKeyframe, (long)keyframeDelay, TimeUnit.MILLISECONDS ); } } } private double getRtt(AbstractEndpoint endpoint) { if (endpoint instanceof Endpoint) { Endpoint localDominantSpeaker = (Endpoint)endpoint; return localDominantSpeaker.getRtt(); } else { // Octo endpoint // TODO(brian): we don't currently have a way to get the RTT from this bridge // to a remote endpoint, so we hard-code a value here. Discussed this with // Boris, and we talked about perhaps having OctoEndpoint periodically // send pings to the remote endpoint to calculate its RTT from the perspective // of this bridge. return 100; } } private double getMaxReceiverRtt(String excludedEndpointId) { return endpointsCache.stream() .filter(ep -> !ep.getID().equalsIgnoreCase(excludedEndpointId)) .map(Endpoint::getRtt) .mapToDouble(Double::valueOf) .max() .orElse(0); } /** * Expires this Conference, its Contents and their * respective Channels. Releases the resources acquired by this * instance throughout its life time and prepares it to be garbage * collected. */ public void expire() { synchronized (this) { if (expired) { return; } else { expired = true; } } logger.info("Expiring."); EventAdmin eventAdmin = getEventAdmin(); if (eventAdmin != null) { eventAdmin.sendEvent(EventFactory.conferenceExpired(this)); } Videobridge videobridge = getVideobridge(); try { videobridge.expireConference(this); } finally { if (logger.isDebugEnabled()) { logger.debug("Expiring endpoints."); } getEndpoints().forEach(AbstractEndpoint::expire); speechActivity.expire(); if (tentacle != null) { tentacle.expire(); tentacle = null; } if (includeInStatistics) { updateStatisticsOnExpire(); } } } /** * Updates the statistics for this conference when it is about to expire. */ private void updateStatisticsOnExpire() { long durationSeconds = Math.round((System.currentTimeMillis() - creationTime) / 1000d); Videobridge.Statistics videobridgeStatistics = getVideobridge().getStatistics(); videobridgeStatistics.totalConferencesCompleted .incrementAndGet(); videobridgeStatistics.totalConferenceSeconds.addAndGet( durationSeconds); videobridgeStatistics.totalBytesReceived.addAndGet( statistics.totalBytesReceived.get()); videobridgeStatistics.totalBytesSent.addAndGet( statistics.totalBytesSent.get()); videobridgeStatistics.totalPacketsReceived.addAndGet( statistics.totalPacketsReceived.get()); videobridgeStatistics.totalPacketsSent.addAndGet( statistics.totalPacketsSent.get()); boolean hasFailed = statistics.hasIceFailedEndpoint && !statistics.hasIceSucceededEndpoint; boolean hasPartiallyFailed = statistics.hasIceFailedEndpoint && statistics.hasIceSucceededEndpoint; if (hasPartiallyFailed) { videobridgeStatistics.totalPartiallyFailedConferences .incrementAndGet(); } if (hasFailed) { videobridgeStatistics.totalFailedConferences.incrementAndGet(); } if (logger.isInfoEnabled()) { StringBuilder sb = new StringBuilder("expire_conf,"); sb.append("duration=").append(durationSeconds) .append(",has_failed=").append(hasFailed) .append(",has_partially_failed=").append(hasPartiallyFailed); logger.info(sb.toString()); } } /** * Finds an Endpoint of this Conference which sends an RTP * stream with a specific SSRC and with a specific MediaType. * * @param receiveSSRC the SSRC of an RTP stream received by this * Conference whose sending Endpoint is to be found * @return Endpoint of this Conference which sends an RTP * stream with the specified ssrc and with the specified * mediaType; otherwise, null */ AbstractEndpoint findEndpointByReceiveSSRC(long receiveSSRC) { return getEndpoints().stream() .filter(ep -> ep.receivesSsrc(receiveSSRC)) .findFirst() .orElse(null); } /** * Returns the OSGi BundleContext in which this Conference is * executing. * * @return the OSGi BundleContext in which the Conference is * executing. */ public BundleContext getBundleContext() { return getVideobridge().getBundleContext(); } /** * Gets an Endpoint participating in this Conference which * has a specific identifier/ID. * * @param id the identifier/ID of the Endpoint which is to be * returned * @return an Endpoint participating in this Conference * which has the specified id or null */ @Nullable public AbstractEndpoint getEndpoint(@NotNull String id) { return endpoints.get( Objects.requireNonNull(id, "id must be non null")); } /** * Initializes a new Endpoint instance with the specified * id and adds it to the list of Endpoints participating * in this Conference. * @param id the identifier/ID of the Endpoint which will be * created * @param iceControlling {@code true} if the ICE agent of this endpoint's * transport will be initialized to serve as a controlling ICE agent; * otherwise, {@code false} * @return an Endpoint participating in this Conference */ @NotNull public Endpoint createLocalEndpoint(String id, boolean iceControlling) throws IOException { final AbstractEndpoint existingEndpoint = getEndpoint(id); if (existingEndpoint instanceof OctoEndpoint) { // It is possible that an Endpoint was migrated from another bridge // in the conference to this one, and the sources lists (which // implicitly signal the Octo endpoints in the conference) haven't // been updated yet. We'll force the Octo endpoint to expire and // we'll continue with the creation of a new local Endpoint for the // participant. existingEndpoint.expire(); } else if (existingEndpoint != null) { throw new IllegalArgumentException("Local endpoint with ID = " + id + "already created"); } final Endpoint endpoint = new Endpoint( id, this, logger, iceControlling); // The propertyChangeListener will weakly reference this // Conference and will unregister itself from the endpoint // sooner or later. endpoint.addPropertyChangeListener(propertyChangeListener); addEndpoint(endpoint); EventAdmin eventAdmin = getEventAdmin(); if (eventAdmin != null) { eventAdmin.sendEvent( EventFactory.endpointCreated(endpoint)); } return endpoint; } /** * An endpoint was added or removed. */ private void endpointsChanged() { speechActivity.endpointsChanged(); } /** * The media stream tracks of one of the endpoints in this conference * changed. * * @param endpoint the endpoint, or {@code null} if it was an Octo endpoint. */ public void endpointTracksChanged(AbstractEndpoint endpoint) { List endpoints = speechActivity.getEndpointIds(); endpointsCache.forEach((e) -> { if (e != endpoint) { e.speechActivityEndpointsChanged(endpoints); } }); } /** * Updates {@link #endpointsCache} with the current contents of * {@link #endpoints}. */ private void updateEndpointsCache() { synchronized (endpointsCacheLock) { ArrayList endpointsList = new ArrayList<>(endpoints.size()); endpoints.values().forEach(e -> { if (e instanceof Endpoint) { endpointsList.add((Endpoint) e); } }); endpointsCache = Collections.unmodifiableList(endpointsList); } } /** * Returns the number of local AND remote {@link Endpoint}s in this {@link Conference}. * * @return the number of local AND remote {@link Endpoint}s in this {@link Conference}. */ public int getEndpointCount() { return endpoints.size(); } /** * Returns the number of local {@link Endpoint}s in this {@link Conference}. * * @return the number of local {@link Endpoint}s in this {@link Conference}. */ public int getLocalEndpointCount() { return getLocalEndpoints().size(); } /** * Gets the Endpoints participating in/contributing to this * Conference. * * @return the Endpoints participating in/contributing to this * Conference */ public List getEndpoints() { return new ArrayList<>(this.endpoints.values()); } /** * Gets the list of endpoints which are local to this bridge (as opposed * to being on a remote bridge through Octo). */ public List getLocalEndpoints() { return endpointsCache; } /** * Gets the JID of the conference focus who has initialized this instance * and from whom requests to manage this instance must come or they will be * ignored. * * @return the JID of the conference focus who has initialized this instance * and from whom requests to manage this instance must come or they will be * ignored */ public final Jid getFocus() { return focus; } /** * Gets the (unique) identifier/ID of this instance. * * @return the (unique) identifier/ID of this instance */ public final String getID() { return id; } /** * Gets the time in milliseconds of the last activity related to this * Conference. * * @return the time in milliseconds of the last activity related to this * Conference */ public long getLastActivityTime() { synchronized (this) { return lastActivityTime; } } /** * Returns the JID of the last known focus. * @return the JID of the last known focus. */ public Jid getLastKnowFocus() { return lastKnownFocus; } /** * Gets an Endpoint participating in this Conference which * has a specific identifier/ID. * * @param id the identifier/ID of the Endpoint which is to be * returned * @return an Endpoint participating in this Conference * or {@code null} if endpoint with specified ID does not exist in a * conference */ @Nullable public Endpoint getLocalEndpoint(String id) { AbstractEndpoint endpoint = getEndpoint(id); if (endpoint instanceof Endpoint) { return (Endpoint) endpoint; } return null; } /** * Gets the speech activity (representation) of the Endpoints of * this Conference. * * @return the speech activity (representation) of the Endpoints of * this Conference */ public ConferenceSpeechActivity getSpeechActivity() { return speechActivity; } /** * Gets the Videobridge which has initialized this * Conference. * * @return the Videobridge which has initialized this * Conference */ public final Videobridge getVideobridge() { return videobridge; } /** * Gets the indicator which determines whether this Conference has * expired. * * @return true if this Conference has expired; otherwise, * false */ public boolean isExpired() { // this.expired starts as 'false' and only ever changes to 'true', // so there is no need to synchronize while reading. return expired; } /** * Notifies this instance that there was a change in the value of a property * of an object in which this instance is interested. * * @param ev a PropertyChangeEvent which specifies the object of * interest, the name of the property and the old and new values of that * property */ @Override @SuppressWarnings("unchecked") public void propertyChange(PropertyChangeEvent ev) { Object source = ev.getSource(); if (isExpired()) { // An expired Conference is to be treated like a null Conference // i.e. it does not handle any PropertyChangeEvents. If possible, // make sure that no further PropertyChangeEvents will be delivered // to this Conference. if (source instanceof PropertyChangeNotifier) { ((PropertyChangeNotifier) source).removePropertyChangeListener( propertyChangeListener); } } else if (Endpoint.SELECTED_ENDPOINTS_PROPERTY_NAME .equals(ev.getPropertyName())) { Set oldSelectedEndpoints = (Set)ev.getOldValue(); Set newSelectedEndpoints = (Set)ev.getNewValue(); // Any endpoints in the oldSelectedEndpoints list which AREN'T // in the newSelectedEndpoints list should have their count decremented oldSelectedEndpoints.stream() .filter( oldSelectedEp -> !newSelectedEndpoints.contains(oldSelectedEp)) .map(this::getEndpoint) .filter(Objects::nonNull) .forEach(AbstractEndpoint::decrementSelectedCount); // Any endpoints in the newSelectedEndpoints list which AREN'T // in the oldSelectedEndpoints list should have their count incremented newSelectedEndpoints.stream() .filter( newSelectedEp -> !oldSelectedEndpoints.contains(newSelectedEp)) .map(this::getEndpoint) .filter(Objects::nonNull) .forEach(AbstractEndpoint::incrementSelectedCount); } } /** * Notifies this conference that one of it's endpoints has expired. * * @param endpoint the Endpoint which expired. */ void endpointExpired(AbstractEndpoint endpoint) { final AbstractEndpoint removedEndpoint; String id = endpoint.getID(); removedEndpoint = endpoints.remove(id); if (removedEndpoint != null) { updateEndpointsCache(); } if (tentacle != null) { tentacle.endpointExpired(id); } if (removedEndpoint != null) { final EventAdmin eventAdmin = getEventAdmin(); if (eventAdmin != null) { eventAdmin.sendEvent( EventFactory.endpointExpired(removedEndpoint)); } endpointsChanged(); } } /** * Adds a specific {@link AbstractEndpoint} instance to the list of * endpoints in this conference. * @param endpoint the endpoint to add. */ public void addEndpoint(AbstractEndpoint endpoint) { if (endpoint.getConference() != this) { throw new IllegalArgumentException("Endpoint belong to other " + "conference = " + endpoint.getConference()); } final AbstractEndpoint replacedEndpoint; replacedEndpoint = endpoints.put(endpoint.getID(), endpoint); updateEndpointsCache(); endpointsChanged(); if (replacedEndpoint != null) { logger.info("Endpoint with id " + endpoint.getID() + ": " + replacedEndpoint + " has been replaced by new " + "endpoint with same id: " + endpoint); } } /** * Notifies this {@link Conference} that one of its {@link Endpoint}s * transport channel has become available. * * @param endpoint the {@link Endpoint} whose transport channel has become * available. */ @Override public void endpointMessageTransportConnected(@NotNull AbstractEndpoint endpoint) { EventAdmin eventAdmin = getEventAdmin(); if (eventAdmin != null) { eventAdmin.postEvent( EventFactory.endpointMessageTransportReady(endpoint)); } if (!isExpired()) { AbstractEndpoint dominantSpeaker = speechActivity.getDominantEndpoint(); if (dominantSpeaker != null) { try { endpoint.sendMessage( createDominantSpeakerEndpointChangeEvent( dominantSpeaker.getID())); } catch (IOException e) { logger.error( "Failed to send dominant speaker update" + " on data channel to " + endpoint.getID(), e); } } } } /** * Sets the JID of the last known focus. * * @param jid the JID of the last known focus. */ public void setLastKnownFocus(Jid jid) { lastKnownFocus = jid; } /** * Notifies this instance that the list of ordered endpoints has changed */ void speechActivityEndpointsChanged(List newEndpointIds) { endpointsCache.forEach( e -> e.speechActivityEndpointsChanged(newEndpointIds)); } /** * Sets the time in milliseconds of the last activity related to this * Conference to the current system time. */ public void touch() { long now = System.currentTimeMillis(); synchronized (this) { if (getLastActivityTime() < now) { lastActivityTime = now; } } } /** * Gets the conference name. * * @return the conference name */ public Localpart getName() { return name; } /** * Returns the EventAdmin instance used by this Conference * and all instances (of {@code Content}, {@code Channel}, etc.) created by * it. * * @return the EventAdmin instance used by this Conference */ public EventAdmin getEventAdmin() { return eventAdmin; } /** * @return the {@link Logger} used by this instance. */ public Logger getLogger() { return logger; } /** * @return the global ID of the conference, or {@code null} if none has been * set. */ public String getGid() { return gid; } /** * {@inheritDoc} *

* @return {@code true} if this {@link Conference} is ready to be expired. */ @Override public boolean shouldExpire() { // Allow a conference to have no endpoints in the first 20 seconds. return getEndpointCount() == 0 && (System.currentTimeMillis() - creationTime > 20000); } /** * {@inheritDoc} */ @Override public void safeExpire() { expireableImpl.safeExpire(); } /** * @return this {@link Conference}'s Colibri shim. */ public ConferenceShim getShim() { return shim; } /** * Broadcasts the packet to all endpoints and tentacles that want it. * * @param packetInfo the packet */ private void sendOut(PacketInfo packetInfo) { String sourceEndpointId = packetInfo.getEndpointId(); // We want to avoid calling 'clone' for the last receiver of this packet // since it's unnecessary. To do so, we'll wait before we clone and send // to an interested handler until after we've determined another handler // is also interested in the packet. We'll give the last handler the // original packet (without cloning). PotentialPacketHandler prevHandler = null; for (Endpoint endpoint : endpointsCache) { if (endpoint.getID().equals(sourceEndpointId)) { continue; } if (endpoint.wants(packetInfo)) { if (prevHandler != null) { prevHandler.send(packetInfo.clone()); } prevHandler = endpoint; } } if (tentacle != null && tentacle.wants(packetInfo)) { if (prevHandler != null) { prevHandler.send(packetInfo.clone()); } prevHandler = tentacle; } if (prevHandler != null) { prevHandler.send(packetInfo); } else { // No one wanted the packet, so the buffer is now free! ByteBufferPool.returnBuffer(packetInfo.getPacket().getBuffer()); } } /** * Gets the audio level listener. */ public AudioLevelListener getAudioLevelListener() { return audioLevelListener; } /** * @return The {@link ConfOctoTransport} for this conference. */ public ConfOctoTransport getTentacle() { if (tentacle == null) { tentacle = new ConfOctoTransport(this); tentacle.addPropertyChangeListener(propertyChangeListener); } return tentacle; } public boolean isOctoEnabled() { return tentacle != null; } /** * Handles an RTP/RTCP packet coming from a specific endpoint. * @param packetInfo */ public void handleIncomingPacket(PacketInfo packetInfo) { Packet packet = packetInfo.getPacket(); if (packet instanceof RtpPacket) { // This is identical to the default 'else' below, but it defined // because the vast majority of packet will follow this path. sendOut(packetInfo); } else if (packet instanceof RtcpFbPliPacket || packet instanceof RtcpFbFirPacket) { long mediaSsrc = (packet instanceof RtcpFbPliPacket) ? ((RtcpFbPliPacket) packet).getMediaSourceSsrc() : ((RtcpFbFirPacket) packet).getMediaSenderSsrc(); // XXX we could make this faster with a map AbstractEndpoint targetEndpoint = findEndpointByReceiveSSRC(mediaSsrc); PotentialPacketHandler pph = null; if (targetEndpoint instanceof Endpoint) { pph = (Endpoint) targetEndpoint; } else if (targetEndpoint instanceof OctoEndpoint) { pph = tentacle; } // This is not a redundant check. With Octo and 3 or more bridges, // some PLI or FIR will come from Octo but the target endpoint will // also be Octo. We need to filter these out. if (pph == null) { if (logger.isDebugEnabled()) { logger.debug("Dropping FIR/PLI for media ssrc " + mediaSsrc); } } else if (pph.wants(packetInfo)) { pph.send(packetInfo); } } else { sendOut(packetInfo); } } /** * Gets a JSON representation of the parts of this object's state that * are deemed useful for debugging. * * @param full if specified the result will include more details and will * include the debug state of the endpoint(s). Otherwise just the IDs and * names of the conference and endpoints are included. * @param endpointId the ID of the endpoint to include. If set to * {@code null}, all endpoints will be included. */ @SuppressWarnings("unchecked") public JSONObject getDebugState(boolean full, String endpointId) { JSONObject debugState = new JSONObject(); debugState.put("id", id); debugState.put("name", name == null ? null : name.toString()); if (full) { debugState.put("gid", gid); debugState.put("expired", expired); debugState.put("creationTime", creationTime); debugState.put("lastActivity", lastActivityTime); debugState.put("speechActivity", speechActivity.getDebugState()); debugState.put("includeInStatistics", includeInStatistics); debugState.put("statistics", statistics.getJson()); //debugState.put("encodingsManager", encodingsManager.getDebugState()); ConfOctoTransport tentacle = this.tentacle; debugState.put( "tentacle", tentacle == null ? null : tentacle.getDebugState()); } JSONObject endpoints = new JSONObject(); debugState.put("endpoints", endpoints); for (Endpoint e : endpointsCache) { if (endpointId == null || endpointId.equals(e.getID())) { endpoints.put(e.getID(), full ? e.getDebugState() : e.getStatsId()); } } return debugState; } /** * Whether this looks like a conference in which the two endpoints are * using a peer-to-peer connection (i.e. none of them are sending audio * or video). * This has false positives when e.g. an endpoint doesn't support p2p * (firefox) and both are audio/video muted. */ public boolean isP2p() { return isInactive() && getEndpointCount() == 2; } /** * Whether the conference is inactive, in the sense that none of its * endpoints are sending audio or video. */ public boolean isInactive() { return getEndpoints().stream().noneMatch(e -> e.isSendingAudio() || e.isSendingVideo()); } /** * Holds conference statistics. */ public static class Statistics { /** * The total number of bytes received in RTP packets in channels in this * conference. Note that this is only updated when channels expire. */ AtomicLong totalBytesReceived = new AtomicLong(); /** * The total number of bytes sent in RTP packets in channels in this * conference. Note that this is only updated when channels expire. */ AtomicLong totalBytesSent = new AtomicLong(); /** * The total number of RTP packets received in channels in this * conference. Note that this is only updated when channels expire. */ AtomicLong totalPacketsReceived = new AtomicLong(); /** * The total number of RTP packets received in channels in this * conference. Note that this is only updated when channels expire. */ AtomicLong totalPacketsSent = new AtomicLong(); /** * Whether at least one endpoint in this conference failed ICE. */ boolean hasIceFailedEndpoint = false; /** * Whether at least one endpoint in this conference completed ICE * successfully. */ boolean hasIceSucceededEndpoint = false; /** * Gets a snapshot of this object's state as JSON. */ @SuppressWarnings("unchecked") private JSONObject getJson() { JSONObject jsonObject = new JSONObject(); jsonObject.put("total_bytes_received", totalBytesReceived.get()); jsonObject.put("total_bytes_sent", totalBytesSent.get()); jsonObject.put("total_packets_received", totalPacketsReceived.get()); jsonObject.put("total_packets_sent", totalPacketsSent.get()); jsonObject.put("has_failed_endpoint", hasIceFailedEndpoint); jsonObject.put("has_succeeded_endpoint", hasIceSucceededEndpoint); return jsonObject; } } /** * This is a no-op diagnostic context (one that will record nothing) meant * to disable logging of time-series for health checks. */ static class NoOpDiagnosticContext extends DiagnosticContext { @Override public TimeSeriesPoint makeTimeSeriesPoint(String timeSeriesName, long tsMs) { return new NoOpTimeSeriesPoint(); } @Override public Object put(@NotNull String key, @NotNull Object value) { return null; } } static class NoOpTimeSeriesPoint extends DiagnosticContext.TimeSeriesPoint { public NoOpTimeSeriesPoint() { this(Collections.emptyMap()); } public NoOpTimeSeriesPoint(Map m) { super(m); } @Override public Object put(String key, Object value) { return null; } } }