Visitors queue

pom.xml

@@ -210,7 +210,7 @@
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>ice4j</artifactId>
-      <version>3.0-59-g71e244d</version>
+      <version>3.0-72-g824cd4b</version>
     </dependency>
     <dependency>
       <groupId>org.opentelecoms.sip</groupId>

src/main/java/org/jitsi/jigasi/CallContext.java

@@ -123,6 +123,11 @@ public class CallContext
      */
     private String authUserId;
 
+    /**
+     * Whether to request visitor when joining.
+     */
+    private boolean requestVisitor = false;
+
     /**
      * Optional bosh url that we use to join a room with the
      * xmpp account.
@@ -619,4 +624,14 @@ public Map<String, String> getExtraHeaders()
     {
         return Collections.unmodifiableMap(this.extraHeaders);
     }
+
+    public boolean isRequestVisitor()
+    {
+        return requestVisitor;
+    }
+
+    public void setRequestVisitor(boolean requestVisitor)
+    {
+        this.requestVisitor = requestVisitor;
+    }
 }

src/main/java/org/jitsi/jigasi/JvbConference.java

@@ -32,6 +32,7 @@
 import org.jitsi.jigasi.stats.*;
 import org.jitsi.jigasi.util.*;
 import org.jitsi.jigasi.version.*;
+import org.jitsi.jigasi.visitor.*;
 import org.jitsi.jigasi.xmpp.extensions.*;
 import org.jitsi.utils.*;
 import org.jitsi.utils.logging.Logger;
@@ -142,6 +143,32 @@ public class JvbConference
      */
     private static final int JVB_ACTIVITY_CHECK_DELAY = 5000;
 
+    /**
+     * The name of the property which enables visitors queue service.
+     */
+    public static final String P_NAME_VISITORS_QUEUE_SERVICE = "org.jitsi.jigasi.VISITOR_QUEUE_SERVICE";
+
+    /**
+     * The visitors queue service url.
+     */
+    private static String visitorsQueueServiceUrl = null;
+    static
+    {
+        visitorsQueueServiceUrl = JigasiBundleActivator.getConfigurationService()
+            .getString(P_NAME_VISITORS_QUEUE_SERVICE);
+    }
+
+    /**
+     * The error code used to indicate that the meeting is not live.
+     * (number that is not clashing with OperationFailedException error codes)
+     */
+    private static final int NOT_LIVE_ERROR_CODE = 101;
+
+    /**
+     * The websocket client to connect to visitors queue if configured.
+     */
+    private WebsocketClient websocketClient;
+
     /**
      * A timer which will be used to schedule a quick non-blocking check whether there is any activity
      * on the bridge side of the call.
@@ -614,6 +641,11 @@ public synchronized void stop()
 
         leaveConferenceRoom();
 
+        if (this.websocketClient != null)
+        {
+            this.websocketClient.disconnect();
+        }
+
         if (jvbCall != null)
         {
             CallManager.hangupCall(jvbCall, true);
@@ -1129,6 +1161,15 @@ public void joinConferenceRoom()
                         }
                     }
                 }
+                else if (opex.getErrorCode() == NOT_LIVE_ERROR_CODE)
+                {
+                    logger.info(this.callContext + " Conference is not live yet.");
+
+                    websocketClient = new WebsocketClient(this, visitorsQueueServiceUrl, this.callContext);
+                    websocketClient.connect();
+
+                    return;
+                }
             }
 
             if (e.getCause() instanceof XMPPException.XMPPErrorException)
@@ -1941,6 +1982,7 @@ public void conferenceMemberRemoved(CallPeerConferenceEvent conferenceEvent)
      * @return Returns vnode if one exist in focus response.
      */
     private String inviteFocus(final EntityBareJid roomIdentifier)
+        throws OperationFailedException
     {
         if (callContext == null || callContext.getRoomJidDomain() == null)
         {
@@ -1955,6 +1997,10 @@ private String inviteFocus(final EntityBareJid roomIdentifier)
         if (JigasiBundleActivator.isSipVisitorsEnabled() && !this.isTranscriber)
         {
             focusInviteIQ.addProperty("visitors-version", "1");
+            if (callContext.isRequestVisitor())
+            {
+                focusInviteIQ.addProperty("visitor", Boolean.TRUE.toString());
+            }
         }
 
         try
@@ -1979,6 +2025,14 @@ private String inviteFocus(final EntityBareJid roomIdentifier)
                 collector = getConnection().createStanzaCollectorAndSend(focusInviteIQ);
                 ConferenceIq res = collector.nextResultOrThrow();
 
+                if (visitorsQueueServiceUrl != null)
+                {
+                    String liveValue = res.getPropertiesMap().get("live");
+                    if (liveValue != null && !Boolean.parseBoolean(liveValue))
+                    {
+                        throw new OperationFailedException("Not live conference", NOT_LIVE_ERROR_CODE);
+                    }
+                }
                 return res.getVnode();
             }
             catch (SmackException

src/main/java/org/jitsi/jigasi/SipGatewaySession.java

@@ -159,6 +159,21 @@ public class SipGatewaySession
     private static final String JITSI_MEET_DOMAIN_TENANT_HEADER_PROPERTY
         = "JITSI_MEET_DOMAIN_TENANT_HEADER_NAME";
 
+    /**
+     * The name of the header to search in the INVITE headers whether to request joining as a visitor.
+     */
+    private final String visitorHeaderName;
+
+    /**
+     * Default value optional INVITE header which specifies whether to join as visitor.
+     */
+    public static final String JITSI_MEET_VISITOR_HEADER_DEFAULT = "Jitsi-Visitor";
+
+    /**
+     * The account property to use to set custom header name for domain tenant.
+     */
+    private static final String JITSI_MEET_VISITOR_HEADER_PROPERTY = "JITSI_MEET_VISITOR_HEADER_NAME";
+
     /**
      * The account property to use to set outbound prefix to be added to all outgoing calls.
      */
@@ -387,6 +402,9 @@ public SipGatewaySession(SipGateway gateway, CallContext callContext)
                 JITSI_MEET_DOMAIN_TENANT_HEADER_PROPERTY,
                 JITSI_MEET_DOMAIN_TENANT_HEADER_DEFAULT);
 
+        visitorHeaderName = sipProvider.getAccountID()
+            .getAccountPropertyString(JITSI_MEET_VISITOR_HEADER_PROPERTY, JITSI_MEET_VISITOR_HEADER_DEFAULT);
+
         heartbeatPeriodInSec = sipProvider.getAccountID()
             .getAccountPropertyInt(HEARTBEAT_SECONDS_PROPERTY, heartbeatPeriodInSec);
 
@@ -747,6 +765,7 @@ public void onJoinJitsiMeetRequest(
                     callContext.setAuthUserId(data.get(authUserIdHeaderName));
                     callContext.setMucAddressPrefix(sipProvider.getAccountID()
                         .getAccountPropertyString(CallContext.MUC_DOMAIN_PREFIX_PROP, null));
+                    callContext.setRequestVisitor(Boolean.parseBoolean(data.get(visitorHeaderName)));
 
                     joinJvbConference(callContext);
                 }

src/main/java/org/jitsi/jigasi/transcription/WhisperWebsocket.java

@@ -17,7 +17,6 @@
  */
 package org.jitsi.jigasi.transcription;
 
-import io.jsonwebtoken.*;
 import org.eclipse.jetty.websocket.api.*;
 import org.eclipse.jetty.websocket.api.annotations.*;
 import org.eclipse.jetty.websocket.client.*;
@@ -29,8 +28,6 @@
 import java.io.*;
 import java.net.*;
 import java.nio.*;
-import java.security.*;
-import java.security.spec.*;
 import java.time.*;
 import java.util.*;
 import java.util.concurrent.*;
@@ -137,29 +134,6 @@ public class WhisperWebsocket
         logger.info("Websocket transcription streaming endpoint: " + websocketUrlConfig);
     }
 
-    private String getJWT() throws NoSuchAlgorithmException, InvalidKeySpecException, IOException
-    {
-        if (privateKey.isEmpty() || privateKeyName.isEmpty())
-        {
-            throw new IOException("Failed generating JWT for Whisper. Missing private key or key name.");
-        }
-        long nowMillis = System.currentTimeMillis();
-        Date now = new Date(nowMillis);
-        KeyFactory kf = KeyFactory.getInstance("RSA");
-        PKCS8EncodedKeySpec keySpecPKCS8 = new PKCS8EncodedKeySpec(Base64.getDecoder().decode(privateKey));
-        PrivateKey finalPrivateKey = kf.generatePrivate(keySpecPKCS8);
-        JwtBuilder builder = Jwts.builder()
-                .setHeaderParam("kid", privateKeyName)
-                .setIssuedAt(now)
-                .setAudience(jwtAudience)
-                .setIssuer("jigasi")
-                .signWith(finalPrivateKey, SignatureAlgorithm.RS256);
-        long expires = nowMillis + (60 * 5 * 1000);
-        Date expiry = new Date(expires);
-        builder.setExpiration(expiry);
-        return builder.compact();
-    }
-
     /**
      * Creates a connection url by concatenating the websocket
      * url with the Connection Id;
@@ -192,7 +166,8 @@ void connect()
                 generateWebsocketUrl();
                 logger.info("Connecting to " + websocketUrl);
                 ClientUpgradeRequest upgradeRequest = new ClientUpgradeRequest();
-                upgradeRequest.setHeader("Authorization", "Bearer " + getJWT());
+                upgradeRequest.setHeader("Authorization", "Bearer " +
+                    org.jitsi.jigasi.util.Util.generateAsapToken(privateKey, privateKeyName, jwtAudience, "jigasi"));
                 ws = new WebSocketClient();
                 ws.start();
                 wsSession = ws.connect(this, new URI(websocketUrl), upgradeRequest).get();

src/main/java/org/jitsi/jigasi/util/Util.java

@@ -17,9 +17,11 @@
  */
 package org.jitsi.jigasi.util;
 
+import io.jsonwebtoken.*;
 import net.java.sip.communicator.impl.protocol.jabber.*;
 import net.java.sip.communicator.service.protocol.*;
 import net.java.sip.communicator.service.protocol.media.*;
+import org.apache.commons.lang3.StringUtils;
 import org.jitsi.jigasi.*;
 import org.jitsi.service.neomedia.*;
 import org.jitsi.service.neomedia.format.*;
@@ -32,7 +34,9 @@
 import org.json.simple.*;
 import org.json.simple.parser.*;
 
+import java.io.*;
 import java.lang.reflect.*;
+import java.security.spec.*;
 import java.util.*;
 
 import java.math.*;
@@ -150,7 +154,7 @@ public static String stringToMD5hash(String toHash)
         }
         catch (NoSuchAlgorithmException e)
         {
-            e.printStackTrace();
+            Logger.getLogger(Util.class).error("Error creating hash", e);
         }
 
         return null;
@@ -339,4 +343,36 @@ public static boolean isJibri(ChatRoomMemberJabberImpl member)
     {
         return checkForFeature(member, JIBRI_FEATURE_NAME);
     }
+
+    /**
+     * Generates asap token.
+     * @return the generated token.
+     */
+    public static String generateAsapToken(
+            String privateKey, String privateKeyId, String audience, String issuer)
+        throws NoSuchAlgorithmException,
+               InvalidKeySpecException,
+               IOException
+    {
+        if (StringUtils.isEmpty(privateKey) || StringUtils.isEmpty(privateKeyId))
+        {
+            throw new IOException("Failed generating JWT for Whisper. Missing private key or key name.");
+        }
+
+        long nowMillis = System.currentTimeMillis();
+        Date now = new Date(nowMillis);
+        KeyFactory kf = KeyFactory.getInstance("RSA");
+        PKCS8EncodedKeySpec keySpecPKCS8 = new PKCS8EncodedKeySpec(Base64.getDecoder().decode(privateKey));
+        PrivateKey finalPrivateKey = kf.generatePrivate(keySpecPKCS8);
+
+        JwtBuilder builder = Jwts.builder()
+            .setHeaderParam("kid", privateKeyId)
+            .setIssuedAt(now)
+            .setAudience(audience)
+            .setIssuer(issuer)
+            .signWith(finalPrivateKey, SignatureAlgorithm.RS256);
+        builder.setExpiration(new Date(nowMillis + (60 * 5 * 1000)));
+
+        return builder.compact();
+    }
 }

src/main/java/org/jitsi/jigasi/visitor/StompUtils.java

@@ -0,0 +1,75 @@
+/*
+ * Jigasi, the JItsi GAteway to SIP.
+ *
+ * Copyright @ 2018 - 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.jigasi.visitor;
+
+import java.nio.*;
+
+/**
+ * The utils for sending/receiving STOMP messages.
+ */
+public class StompUtils
+{
+    final static String NEW_LINE = "\n";
+    final static String END = "\u0000";
+    final static String EMPTY_LINE = "";
+    final static String DELIMITER = ":";
+    final static ByteBuffer PING_BODY = ByteBuffer.wrap(new byte[] {'\n'});
+
+    private static String buildHeader(String key, String value)
+    {
+        if (value != null)
+        {
+            return key + ':' + value + NEW_LINE;
+        }
+        else
+        {
+            return key + NEW_LINE;
+        }
+    }
+
+    /**
+     * Builds the connect message to send.
+     * @param token The token to authenticate.
+     * @param heartbeatOutgoing The ms to send for outgoing heartbeat interval.
+     * @param heartbeatIncoming The ms to send for incoming heartbeat interval.
+     * @return The message.
+     */
+    static String buildConnectMessage(String token, long heartbeatOutgoing, long heartbeatIncoming)
+    {
+        String headers = buildHeader("CONNECT", null);
+        headers += buildHeader("accept-version", "1.2,1.1,1.0");
+        headers += buildHeader("heart-beat", heartbeatOutgoing + "," + heartbeatIncoming);
+        headers += buildHeader("Authorization", "Bearer " + token);
+
+        return headers + NEW_LINE + END;
+    }
+
+    /**
+     * Builds a subscribe message.
+     * @param topic The topic to subscribe to.
+     * @return The message.
+     */
+    static String buildSubscribeMessage(String topic)
+    {
+        String headers = buildHeader("SUBSCRIBE", null);
+        headers += buildHeader("destination", topic);
+        headers += buildHeader("id", "1"); // this is the first and only message we send
+
+        return headers + NEW_LINE + END;
+    }
+}