<commit_before>package org.asteriskjava.pbx.internal.activity;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;

import org.asteriskjava.pbx.ActivityCallback;
import org.asteriskjava.pbx.AsteriskSettings;
import org.asteriskjava.pbx.Call;
import org.asteriskjava.pbx.CallImpl;
import org.asteriskjava.pbx.Channel;
import org.asteriskjava.pbx.EndPoint;
import org.asteriskjava.pbx.ListenerPriority;
import org.asteriskjava.pbx.PBXException;
import org.asteriskjava.pbx.PBXFactory;
import org.asteriskjava.pbx.activities.SplitActivity;
import org.asteriskjava.pbx.agi.AgiChannelActivityHold;
import org.asteriskjava.pbx.asterisk.wrap.actions.RedirectAction;
import org.asteriskjava.pbx.asterisk.wrap.events.ManagerEvent;
import org.asteriskjava.pbx.internal.core.AsteriskPBX;
import org.asteriskjava.pbx.internal.core.ChannelProxy;
import org.asteriskjava.util.Log;
import org.asteriskjava.util.LogFactory;

/**
 * The SplitActivity is used by the AsteriksPBX to split a call and place the
 * component channels into the specialist Activity fastagi. The SplitActivity
 * can only split a call with two channels. The (obvious?) limitation is that we
 * can't split something in a conference call as it has more than two channels.
 * 
 * @author bsutton
 */
public class SplitActivityImpl extends ActivityHelper<SplitActivity> implements SplitActivity
{
    private static final Log logger = LogFactory.getLog(SplitActivityImpl.class);

    private Call _callToSplit;

    private Channel channel1;

    private Channel channel2;

    private Call _lhsCall;

    private Call _rhsCall;

    /**
     * Splits a call by moving each of its two channels into the Activity agi.
     * The channels will sit in the agi (with no audio) until something is done
     * with them. As such you should leave them split for too long.
     * 
     * @param callToSplit The call to split
     * @param listener
     */
    public SplitActivityImpl(final Call callToSplit, final ActivityCallback<SplitActivity> listener)
    {
        super("SplitActivity", listener);

        this._callToSplit = callToSplit;

        List<Channel> tmp = callToSplit.getChannels();
        if (tmp.size() > 1)
        {
            channel1 = tmp.get(0);
            channel2 = tmp.get(1);
        }
        this.startActivity(true);

    }

    @Override
    public boolean doActivity() throws PBXException
    {

        SplitActivityImpl.logger.debug("*******************************************************************************");
        SplitActivityImpl.logger.info("***********                    begin split               ****************");
        SplitActivityImpl.logger.info("***********            " + this.channel1 + "                 ****************");
        SplitActivityImpl.logger.debug("***********            " + this.channel2 + "                 ****************");
        SplitActivityImpl.logger.debug("*******************************************************************************");

        // Splits the originating and secondary channels by moving each of them
        // into the associated
        // target.
        boolean success = false;

        if (this.channel2 != null)
        {
            success = splitTwo();

            // Now update the call to reflect the split
            if (success)
            {
                this._lhsCall = ((CallImpl) this._callToSplit).split(channel1);
                this._rhsCall = ((CallImpl) this._callToSplit).split(channel2);
            }
        }
        return success;
    }

    @Override
    public HashSet<Class< ? extends ManagerEvent>> requiredEvents()
    {
        HashSet<Class< ? extends ManagerEvent>> required = new HashSet<>();

        // No events required.
        return required;
    }

    @Override
    synchronized public void onManagerEvent(final ManagerEvent event)
    {
        // NOOP
    }

    @Override
    public ListenerPriority getPriority()
    {
        return ListenerPriority.NORMAL;
    }

    /**
     * After a call has been split we get two new calls. One will hold the
     * original remote party and the other will hold the original local party.
     * 
     * @return the call which holds the original remote party.
     */
    @Override
    public Call getLHSCall()
    {
        return this._lhsCall;
    }

    /**
     * After a call has been split we get two new calls. One will hold the
     * original remote party and the other will hold the original local party.
     * 
     * @return the call which holds the original local party.
     */
    @Override
    public Call getRHSCall()
    {
        return this._rhsCall;
    }

    /**
     * Splits two channels moving them to defined endpoints.
     * 
     * @param lhs
     * @param lhsTarget
     * @param lhsTargetContext
     * @param rhs
     * @param rhsTarget
     * @param rhsTargetContext
     * @return
     * @throws PBXException
     */
    private boolean splitTwo() throws PBXException
    {
        final AsteriskSettings profile = PBXFactory.getActiveProfile();
        AsteriskPBX pbx = (AsteriskPBX) PBXFactory.getActivePBX();

        if (channel1 == channel2)
        {
            throw new NullPointerException(
                    "channel1 is the same as channel2. if I let this happen, asterisk will core dump :)");
        }

        List<Channel> channels = new LinkedList<>();
        channels.add(channel1);
        channels.add(channel2);
        if (!pbx.waitForChannelsToQuiescent(channels, 3000))
        {
            logger.error(callSite, callSite);
            throw new PBXException(
                    "Channel: " + channel1 + " or " + channel2 + " cannot be split as they are still in transition.");
        }

        /*
         * redirects the specified channels to the specified endpoints. Returns
         * true or false reflecting success.
         */

        AgiChannelActivityHold agi1 = new AgiChannelActivityHold();
        AgiChannelActivityHold agi2 = new AgiChannelActivityHold();

        pbx.setVariable(channel1, "proxyId", "" + ((ChannelProxy) channel1).getIdentity());
        pbx.setVariable(channel2, "proxyId", "" + ((ChannelProxy) channel2).getIdentity());

        channel1.setCurrentActivityAction(agi1);
        channel2.setCurrentActivityAction(agi2);

        final String agiExten = profile.getAgiExtension();
        final String agiContext = profile.getManagementContext();
        logger.debug("splitTwo channel lhs:" + channel1 + " to " + agiExten + " in context " + agiContext + " from "
                + this._callToSplit);

        final EndPoint extensionAgi = pbx.getExtensionAgi();
        final RedirectAction redirect = new RedirectAction(channel1, agiContext, extensionAgi, 1);
        redirect.setExtraChannel(channel2);
        redirect.setExtraContext(agiContext);
        redirect.setExtraExten(extensionAgi);
        redirect.setExtraPriority(1);
        // logger.error(redirect);

        boolean ret = false;
        {
            try
            {

                // final ManagerResponse response =
                pbx.sendAction(redirect, 1000);
                double ctr = 0;
                while ((!agi1.hasCallReachedAgi() || !agi2.hasCallReachedAgi()) && ctr < 10)
                {
                    Thread.sleep(100);
                    ctr += 100.0 / 1000.0;
                    if (!agi1.hasCallReachedAgi())
                    {
                        logger.info("Waiting on (agi1) " + channel1);
                    }
                    if (!agi2.hasCallReachedAgi())
                    {
                        logger.info("Waiting on (agi2) " + channel2);
                    }
                }
                ret = agi1.hasCallReachedAgi() && agi2.hasCallReachedAgi();

            }
            catch (final Exception e)
            {
                logger.error(e, e);
            }<commit_msg>fix for SplitActivityImpl to ensure that the channels finish transitioning before returning<commit_after>package org.asteriskjava.pbx.internal.activity;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.asteriskjava.pbx.ActivityCallback;
import org.asteriskjava.pbx.AsteriskSettings;
import org.asteriskjava.pbx.Call;
import org.asteriskjava.pbx.CallImpl;
import org.asteriskjava.pbx.Channel;
import org.asteriskjava.pbx.EndPoint;
import org.asteriskjava.pbx.ListenerPriority;
import org.asteriskjava.pbx.PBXException;
import org.asteriskjava.pbx.PBXFactory;
import org.asteriskjava.pbx.activities.SplitActivity;
import org.asteriskjava.pbx.agi.AgiChannelActivityHold;
import org.asteriskjava.pbx.asterisk.wrap.actions.RedirectAction;
import org.asteriskjava.pbx.asterisk.wrap.events.BridgeEvent;
import org.asteriskjava.pbx.asterisk.wrap.events.ManagerEvent;
import org.asteriskjava.pbx.internal.core.AsteriskPBX;
import org.asteriskjava.pbx.internal.core.ChannelProxy;
import org.asteriskjava.util.Log;
import org.asteriskjava.util.LogFactory;

/**
 * The SplitActivity is used by the AsteriksPBX to split a call and place the
 * component channels into the specialist Activity fastagi. The SplitActivity
 * can only split a call with two channels. The (obvious?) limitation is that we
 * can't split something in a conference call as it has more than two channels.
 * 
 * @author bsutton
 */
public class SplitActivityImpl extends ActivityHelper<SplitActivity> implements SplitActivity
{
    private static final Log logger = LogFactory.getLog(SplitActivityImpl.class);

    private Call _callToSplit;

    private Channel channel1;

    private Channel channel2;

    private Call _lhsCall;

    private Call _rhsCall;

    /**
     * Splits a call by moving each of its two channels into the Activity agi.
     * The channels will sit in the agi (with no audio) until something is done
     * with them. As such you should leave them split for too long.
     * 
     * @param callToSplit The call to split
     * @param listener
     */
    public SplitActivityImpl(final Call callToSplit, final ActivityCallback<SplitActivity> listener)
    {
        super("SplitActivity", listener);

        this._callToSplit = callToSplit;

        List<Channel> tmp = callToSplit.getChannels();
        if (tmp.size() > 1)
        {
            channel1 = tmp.get(0);
            channel2 = tmp.get(1);
        }
        this.startActivity(true);

    }

    @Override
    public boolean doActivity() throws PBXException
    {

        SplitActivityImpl.logger.debug("*******************************************************************************");
        SplitActivityImpl.logger.info("***********                    begin split               ****************");
        SplitActivityImpl.logger.info("***********            " + this.channel1 + "                 ****************");
        SplitActivityImpl.logger.debug("***********            " + this.channel2 + "                 ****************");
        SplitActivityImpl.logger.debug("*******************************************************************************");

        // Splits the originating and secondary channels by moving each of them
        // into the associated
        // target.
        boolean success = false;

        if (this.channel2 != null)
        {
            success = splitTwo();

            // Now update the call to reflect the split
            if (success)
            {
                this._lhsCall = ((CallImpl) this._callToSplit).split(channel1);
                this._rhsCall = ((CallImpl) this._callToSplit).split(channel2);
            }
        }
        return success;
    }

    @Override
    public HashSet<Class< ? extends ManagerEvent>> requiredEvents()
    {
        HashSet<Class< ? extends ManagerEvent>> required = new HashSet<>();

        required.add(BridgeEvent.class);
        return required;
    }

    final CountDownLatch unlinkRecieved = new CountDownLatch(1);

    @Override
    synchronized public void onManagerEvent(final ManagerEvent event)
    {

        if (event instanceof BridgeEvent)
        {
            BridgeEvent link = (BridgeEvent) event;
            if (link.isUnlink())
            {
                if (link.getChannel1().isSame(channel1) || link.getChannel1().isSame(channel2))
                {
                    unlinkRecieved.countDown();
                }
            }
        }
    }

    @Override
    public ListenerPriority getPriority()
    {
        return ListenerPriority.NORMAL;
    }

    /**
     * After a call has been split we get two new calls. One will hold the
     * original remote party and the other will hold the original local party.
     * 
     * @return the call which holds the original remote party.
     */
    @Override
    public Call getLHSCall()
    {
        return this._lhsCall;
    }

    /**
     * After a call has been split we get two new calls. One will hold the
     * original remote party and the other will hold the original local party.
     * 
     * @return the call which holds the original local party.
     */
    @Override
    public Call getRHSCall()
    {
        return this._rhsCall;
    }

    /**
     * Splits two channels moving them to defined endpoints.
     * 
     * @param lhs
     * @param lhsTarget
     * @param lhsTargetContext
     * @param rhs
     * @param rhsTarget
     * @param rhsTargetContext
     * @return
     * @throws PBXException
     */
    private boolean splitTwo() throws PBXException
    {
        final AsteriskSettings profile = PBXFactory.getActiveProfile();
        AsteriskPBX pbx = (AsteriskPBX) PBXFactory.getActivePBX();

        if (channel1 == channel2)
        {
            throw new NullPointerException(
                    "channel1 is the same as channel2. if I let this happen, asterisk will core dump :)");
        }

        List<Channel> channels = new LinkedList<>();
        channels.add(channel1);
        channels.add(channel2);
        if (!pbx.waitForChannelsToQuiescent(channels, 3000))
        {
            logger.error(callSite, callSite);
            throw new PBXException(
                    "Channel: " + channel1 + " or " + channel2 + " cannot be split as they are still in transition.");
        }

        /*
         * redirects the specified channels to the specified endpoints. Returns
         * true or false reflecting success.
         */

        AgiChannelActivityHold agi1 = new AgiChannelActivityHold();
        AgiChannelActivityHold agi2 = new AgiChannelActivityHold();

        pbx.setVariable(channel1, "proxyId", "" + ((ChannelProxy) channel1).getIdentity());
        pbx.setVariable(channel2, "proxyId", "" + ((ChannelProxy) channel2).getIdentity());

        channel1.setCurrentActivityAction(agi1);
        channel2.setCurrentActivityAction(agi2);

        final String agiExten = profile.getAgiExtension();
        final String agiContext = profile.getManagementContext();
        logger.debug("splitTwo channel lhs:" + channel1 + " to " + agiExten + " in context " + agiContext + " from "
                + this._callToSplit);

        final EndPoint extensionAgi = pbx.getExtensionAgi();
        final RedirectAction redirect = new RedirectAction(channel1, agiContext, extensionAgi, 1);
        redirect.setExtraChannel(channel2);
        redirect.setExtraContext(agiContext);
        redirect.setExtraExten(extensionAgi);
        redirect.setExtraPriority(1);
        // logger.error(redirect);

        boolean ret = false;
        {
            try
            {

                // final ManagerResponse response =
                pbx.sendAction(redirect, 1000);

                // wait for channels to unbridge
                if (!unlinkRecieved.await(5000, TimeUnit.MILLISECONDS))
                {
                    logger.error("There was no unlink event");
                }
                else
                {
                    logger.warn("Channels are unlinked, now waiting for Quiescent");
                }

                channels.clear();

                channels.add(channel1);
                channels.add(channel2);
                if (!pbx.waitForChannelsToQuiescent(channels, 3000))
                {
                    logger.error(callSite, callSite);
                    throw new PBXException("Channel: " + channel1 + " or " + channel2
                            + " cannot be split as they are still in transition.");
                }

                double ctr = 0;
                while ((!agi1.hasCallReachedAgi() || !agi2.hasCallReachedAgi()) && ctr < 10)
                {
                    Thread.sleep(100);
                    ctr += 100.0 / 1000.0;
                    if (!agi1.hasCallReachedAgi())
                    {
                        logger.info("Waiting on (agi1) " + channel1);
                    }
                    if (!agi2.hasCallReachedAgi())
                    {
                        logger.info("Waiting on (agi2) " + channel2);
                    }
                }
                ret = agi1.hasCallReachedAgi() && agi2.hasCallReachedAgi();

            }
            catch (final Exception e)
            {
                logger.error(e, e);
            }
{'fraction_non_alphanumeric': 0.11492390486661652, 'fraction_numerical': 0.008853654302413055, 'mean_word_length': 2.1669415429723293, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6002328', 'n_tokens_mistral': 4736, 'n_tokens_neox': 4443, 'n_words': 1196}