<commit_before>/*
 * Copyright @ 2015 Atlassian Pty Ltd
 *
 * 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.pubsub;

import java.util.*;
import java.util.concurrent.*;

import net.java.sip.communicator.util.*;

import org.jitsi.videobridge.stats.*;
import org.jitsi.videobridge.xmpp.*;
import org.jivesoftware.smack.packet.*;
import org.jivesoftware.smackx.pubsub.*;
import org.jivesoftware.smackx.pubsub.packet.*;
import org.osgi.framework.*;

/**
 * Implements some parts of PubSub (XEP-0060: Publish-Subscribe) for the
 * purposes of a publisher (e.g. statistics transport).
 *
 * @author <NAME>
 * @author <NAME>
 */
public class PubSubPublisher
{
    /**
     * Maps a service name (e.g. &quot;pubsub.example.com&quot;) to the
     * <tt>PubSubPublisher</tt> instance responsible for it.
     */
    private static final Map<String, PubSubPublisher> instances
        = new ConcurrentHashMap<>();

    /**
     * The <tt>Logger</tt> used by the <tt>PubSubPublisher</tt> class and its
     * instances to print debug information.
     */
    private static final Logger logger
        = Logger.getLogger(PubSubPublisher.class);

    /**
     * The default timeout of the packets in milliseconds.
     */
    private static final int PACKET_TIMEOUT = 5000;

    /**
     * Gets a <tt>PubSubPublisher</tt> instance for a specific service (name).
     * If a <tt>PubSubPublisher</tt> instance for the specified
     * <tt>serviceName</tt> does not exist yet, a new instance is initialized.
     *
     * @param serviceName the name of the service
     * @return the <tt>PubSubPublisher</tt> instance for the specified
     * <tt>serviceName</tt>
     */
    public static PubSubPublisher getPubsubManager(String serviceName)
    {
        PubSubPublisher publisher = instances.get(serviceName);

        if (publisher == null)
        {
            publisher = new PubSubPublisher(serviceName);
            instances.put(serviceName, publisher);
        }

        return publisher;
    }

    /**
     * Handles received response IQ packet.
     *
     * @param response the IQ packet.
     */
    public static void handleIQResponse(IQ response)
    {
        IQ.Type type = response.getType();

        if (IQ.Type.ERROR.equals(type))
        {
            PubSubPublisher publisher = instances.get(response.getFrom());

            if (publisher != null)
            {
                publisher.handleErrorResponse(response);
            }
        }
        else if (IQ.Type.RESULT.equals(type))
        {
            PubSubPublisher publisher = instances.get(response.getFrom());

            if (publisher != null)
            {
                publisher.handleCreateNodeResponse(response);
                publisher.handleConfigureResponse(response);
                publisher.handlePublishResponse(response);
            }
        }
    }

    /**
     * Releases the resources for the <tt>PubSubPublisher</tt> and removes it
     * from the list of available instances.
     *
     * @param publisher the <tt>PubSubPublisher</tt> release.
     */
    public static void releasePubsubManager(PubSubPublisher publisher)
    {
        instances.values().remove(publisher);
        publisher.dispose();
    }

    /**
     * Listeners for response events.
     */
    private List<PubSubResponseListener> listeners = new LinkedList<>();

    /**
     * List of the accessible PubSub nodes.
     */
    private List<String> nodes = new LinkedList<>();

    /**
     * Map with the requests for configuring a node.
     */
    private Map<String, String> pendingConfigureRequests
        = new ConcurrentHashMap<>();

    /**
     * Map with the requests for node creation.
     */
    private Map<String, String> pendingCreateRequests
        = new ConcurrentHashMap<>();

    /**
     * Map with the publish requests.
     */
    private Map<String, String> pendingPublishRequests
        = new ConcurrentHashMap<>();

    /**
     * The name of the PubSub service.
     */
    private String serviceName;

    /**
     * Timer for timeout of the requests that we are sending.
     */
    private Timer timeoutTimer = new Timer();

    /**
     * Initializes a new <tt>PubSubPublisher</tt> instance for a specific
     * service (name).
     *
     * @param serviceName the name of the service.
     */
    private PubSubPublisher(String serviceName)
    {
        this.serviceName = serviceName;
    }

    /**
     * Adds a new <tt>PubSubResponseListener</tt> to the list of listeners.
     *
     * @param l the listener to add.
     * @throws NullPointerException if <tt>l</tt> is <tt>null</tt>
     */
    public void addResponseListener(PubSubResponseListener l)
    {
        if (l == null)
            throw new NullPointerException("l");
        else if(!listeners.contains(l))
            listeners.add(l);
    }

    /**
     * Configures PubSub node.
     *
     * @param nodeName the name of the node
     */
    private void configureNode(String nodeName)
    {
        ConfigureForm cfg = new ConfigureForm(FormType.submit);
        PubSub pubsub = new PubSub();

        cfg.setAccessModel(AccessModel.open);
        cfg.setPersistentItems(false);
        cfg.setPublishModel(PublishModel.open);
        pubsub.setTo(serviceName);
        pubsub.setType(IQ.Type.SET);

        final String packetID = IQ.nextID();

        pubsub.setPacketID(packetID);
        pubsub.addExtension(
            new FormNode(FormNodeType.CONFIGURE_OWNER, nodeName ,cfg));
        try
        {
            send(pubsub);
        }
        catch (Exception e)
        {
            logger.error("Error sending configuration form.");
            fireResponseCreateEvent(PubSubResponseListener.Response.SUCCESS);
            return;
        }
        pendingConfigureRequests.put(packetID, nodeName);
        timeoutTimer.schedule(
                new TimerTask()
                {
                    @Override
                    public void run()
                    {
                        String nodeName
                            = pendingConfigureRequests.remove(packetID);

                        if(nodeName != null)
                        {
                            logger.error(
                                    "Timed out a configuration request "
                                        + "(packetID=: " + packetID
                                        + " nodeName=" + nodeName + ")");
                            fireResponseCreateEvent(
                                    PubSubResponseListener.Response.SUCCESS);
                        }
                    }
                },
                PACKET_TIMEOUT);
    }

    /**
     * Creates a PubSub node.
     *
     * @param nodeName the name of the node.
     * @throws Exception if sending the request fails.
     */
    public void createNode(String nodeName)
        throws Exception
    {
        PubSub request = new PubSub();

        request.setTo(serviceName);
        request.setType(IQ.Type.SET);

        final String packetID = Packet.nextID();

        request.setPacketID(packetID);
        request.addExtension(
                new NodeExtension(PubSubElementType.CREATE, nodeName));

        pendingCreateRequests.put(packetID, nodeName);
        timeoutTimer.schedule(
                new TimerTask()
                {
                    @Override
                    public void run()
                    {
                        String nodeName = pendingCreateRequests.remove(packetID);
                        if (nodeName != null)
                        {
                            logger.warn("Timed out a create request with ID "
                                            + packetID);
                        }
                    }
                },
                PACKET_TIMEOUT);

        send(request);
    }

    /**
     * Releases the resources of this <tt>PubSubPublisher</tt> i.e. prepares it
     * for garbage collection.
     */
    private void dispose()
    {
        timeoutTimer.cancel();
        timeoutTimer = null;

        listeners = null;
        nodes = null;
        pendingConfigureRequests = null;
        pendingCreateRequests = null;
        pendingPublishRequests = null;
        serviceName = null;
    }

    /**
     * Fires event about the response of creating a node.
     *<commit_msg>fix: Fixes a failure to properly initialize the PubSub publisher.<commit_after>/*
 * Copyright @ 2015 Atlassian Pty Ltd
 *
 * 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.pubsub;

import java.util.*;
import java.util.concurrent.*;

import net.java.sip.communicator.util.*;

import org.jitsi.videobridge.stats.*;
import org.jitsi.videobridge.xmpp.*;
import org.jivesoftware.smack.packet.*;
import org.jivesoftware.smackx.pubsub.*;
import org.jivesoftware.smackx.pubsub.packet.*;
import org.osgi.framework.*;

/**
 * Implements some parts of PubSub (XEP-0060: Publish-Subscribe) for the
 * purposes of a publisher (e.g. statistics transport).
 *
 * @author <NAME>
 * @author <NAME>
 */
public class PubSubPublisher
{
    /**
     * Maps a service name (e.g. &quot;pubsub.example.com&quot;) to the
     * <tt>PubSubPublisher</tt> instance responsible for it.
     */
    private static final Map<String, PubSubPublisher> instances
        = new ConcurrentHashMap<>();

    /**
     * The <tt>Logger</tt> used by the <tt>PubSubPublisher</tt> class and its
     * instances to print debug information.
     */
    private static final Logger logger
        = Logger.getLogger(PubSubPublisher.class);

    /**
     * The default timeout of the packets in milliseconds.
     */
    private static final int PACKET_TIMEOUT = 5000;

    /**
     * Gets a <tt>PubSubPublisher</tt> instance for a specific service (name).
     * If a <tt>PubSubPublisher</tt> instance for the specified
     * <tt>serviceName</tt> does not exist yet, a new instance is initialized.
     *
     * @param serviceName the name of the service
     * @return the <tt>PubSubPublisher</tt> instance for the specified
     * <tt>serviceName</tt>
     */
    public static PubSubPublisher getPubsubManager(String serviceName)
    {
        PubSubPublisher publisher = instances.get(serviceName);

        if (publisher == null)
        {
            publisher = new PubSubPublisher(serviceName);
            instances.put(serviceName, publisher);
        }

        return publisher;
    }

    /**
     * Handles received response IQ packet.
     *
     * @param response the IQ packet.
     */
    public static void handleIQResponse(IQ response)
    {
        IQ.Type type = response.getType();

        if (IQ.Type.ERROR.equals(type))
        {
            PubSubPublisher publisher = instances.get(response.getFrom());

            if (publisher != null)
            {
                publisher.handleErrorResponse(response);
            }
        }
        else if (IQ.Type.RESULT.equals(type))
        {
            PubSubPublisher publisher = instances.get(response.getFrom());

            if (publisher != null)
            {
                publisher.handleCreateNodeResponse(response);
                publisher.handleConfigureResponse(response);
                publisher.handlePublishResponse(response);
            }
        }
    }

    /**
     * Releases the resources for the <tt>PubSubPublisher</tt> and removes it
     * from the list of available instances.
     *
     * @param publisher the <tt>PubSubPublisher</tt> release.
     */
    public static void releasePubsubManager(PubSubPublisher publisher)
    {
        instances.values().remove(publisher);
        publisher.dispose();
    }

    /**
     * Listeners for response events.
     */
    private List<PubSubResponseListener> listeners = new LinkedList<>();

    /**
     * List of the accessible PubSub nodes.
     */
    private List<String> nodes = new LinkedList<>();

    /**
     * Map with the requests for configuring a node.
     */
    private Map<String, String> pendingConfigureRequests
        = new ConcurrentHashMap<>();

    /**
     * Map with the requests for node creation.
     */
    private Map<String, String> pendingCreateRequests
        = new ConcurrentHashMap<>();

    /**
     * Map with the publish requests.
     */
    private Map<String, String> pendingPublishRequests
        = new ConcurrentHashMap<>();

    /**
     * The name of the PubSub service.
     */
    private String serviceName;

    /**
     * Timer for timeout of the requests that we are sending.
     */
    private Timer timeoutTimer = new Timer();

    /**
     * Initializes a new <tt>PubSubPublisher</tt> instance for a specific
     * service (name).
     *
     * @param serviceName the name of the service.
     */
    private PubSubPublisher(String serviceName)
    {
        this.serviceName = serviceName;
    }

    /**
     * Adds a new <tt>PubSubResponseListener</tt> to the list of listeners.
     *
     * @param l the listener to add.
     * @throws NullPointerException if <tt>l</tt> is <tt>null</tt>
     */
    public void addResponseListener(PubSubResponseListener l)
    {
        if (l == null)
            throw new NullPointerException("l");
        else if(!listeners.contains(l))
            listeners.add(l);
    }

    /**
     * Configures PubSub node.
     *
     * @param nodeName the name of the node
     */
    private void configureNode(String nodeName)
    {
        ConfigureForm cfg = new ConfigureForm(FormType.submit);
        PubSub pubsub = new PubSub();

        cfg.setAccessModel(AccessModel.open);
        cfg.setPersistentItems(false);
        cfg.setPublishModel(PublishModel.open);
        pubsub.setTo(serviceName);
        pubsub.setType(IQ.Type.SET);

        final String packetID = IQ.nextID();

        pubsub.setPacketID(packetID);
        pubsub.addExtension(
            new FormNode(FormNodeType.CONFIGURE_OWNER, nodeName ,cfg));
        try
        {
            send(pubsub);
        }
        catch (Exception e)
        {
            logger.error("Error sending configuration form.");
            fireResponseCreateEvent(PubSubResponseListener.Response.SUCCESS);
            return;
        }
        pendingConfigureRequests.put(packetID, nodeName);
        timeoutTimer.schedule(
                new TimerTask()
                {
                    @Override
                    public void run()
                    {
                        String nodeName
                            = pendingConfigureRequests.remove(packetID);

                        if(nodeName != null)
                        {
                            logger.error(
                                    "Timed out a configuration request "
                                        + "(packetID=: " + packetID
                                        + " nodeName=" + nodeName + ")");
                            fireResponseCreateEvent(
                                    PubSubResponseListener.Response.SUCCESS);
                        }
                    }
                },
                PACKET_TIMEOUT);
    }

    /**
     * Creates a PubSub node.
     *
     * @param nodeName the name of the node.
     * @throws Exception if sending the request fails.
     */
    public void createNode(String nodeName)
        throws Exception
    {
        PubSub request = new PubSub();

        request.setTo(serviceName);
        request.setType(IQ.Type.SET);

        final String packetID = Packet.nextID();

        request.setPacketID(packetID);
        request.addExtension(
                new NodeExtension(PubSubElementType.CREATE, nodeName));

        pendingCreateRequests.put(packetID, nodeName);

        // Send the request before starting the timer, as we have observed
        // sending to be significantly delayed (possibly waiting for the XMPP
        // component connection to become ready).
        send(request);

        timeoutTimer.schedule(
                new TimerTask()
                {
                    @Override
                    public void run()
                    {
                        String nodeName = pendingCreateRequests.remove(packetID);
                        if (nodeName != null)
                        {
                            logger.warn("Timed out a create request with ID "
                                            + packetID);
                        }
                    }
                },
                PACKET_TIMEOUT);
    }

    /**
     * Releases the resources of this <tt>PubSubPublisher</tt> i.e. prepares it
     * for garbage collection.
     */
    private void dispose()
    {
        timeoutTimer.cancel();
        timeoutTimer = null;

        listeners = null;
        nodes = null;
        pendingConfigureRequests = null;
        pendingCreateRequests = null;
        pendingPublishRequests = null;
        serviceName = null;
    }

    /**
     * Fires event about the response of creating a node.
     *
{'fraction_non_alphanumeric': 0.09303874764072388, 'fraction_numerical': 0.0017763961363384034, 'mean_word_length': 1.8325471698113207, 'pattern_counts': {'":': 0, '<': 91, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3401924', 'n_tokens_mistral': 4799, 'n_tokens_neox': 4568, 'n_words': 1281}