<commit_before>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.feedapi;

import com.yahoo.concurrent.SystemTimer;
import com.yahoo.jdisc.Metric;
import com.yahoo.log.LogLevel;
import com.yahoo.messagebus.*;
import com.yahoo.clientmetrics.RouteMetricSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

/**
 * This class allows multiple clients to use one shared messagebus session.
 * The user should create a ResultCallback, which acts as a "session" for that
 * client, and send one or more messages using the send() methods.
 * When done sending messages, the client can wait for all messages to be replied to
 * using the waitForPending method.
 */
public class SharedSender implements ReplyHandler {

    public static final Logger log = Logger.getLogger(SharedSender.class.getName());

    private SendSession sender;
    private RouteMetricSet metrics;
    private static final int REACT_LATENCY_ON_RACE = 5;
    private static final int REACT_LATENCY_ON_WATERMARK = 5;

    private ConcurrentHashMap<ResultCallback, OwnerState> activeOwners = new ConcurrentHashMap<>();

    /**
     * Creates a new shared sender.
     * If oldsender != null, we copy that status information from that sender.
     */
    public SharedSender(String route, SessionFactory factory, SharedSender oldSender, Metric metric) {
        if (factory != null) {
            sender = factory.createSendSession(this, metric);
        }

        if (oldSender != null) {
            this.metrics = oldSender.metrics;
        } else {
            metrics = new RouteMetricSet(route, null);
        }
    }

    public RouteMetricSet getMetrics() {
        return metrics;
    }

    public void remove(ResultCallback owner) {
        OwnerState state = activeOwners.remove(owner);
        if (state != null) {
            state.clearPending();
        }
    }

    public void shutdown() {
        try {
            while ( ! activeOwners.isEmpty()) {
                Thread.sleep(10);
            }
        } catch (InterruptedException e) {
        }
        sender.close();
    }

    /**
     * Waits until there are no more pending documents
     * for the given callback, or the timeout expires.
     *
     * @param owner     The callback to check for.
     * @param timeoutMs The number of milliseconds to wait, or -1 to wait indefinitely.
     * @return true if there were no more pending, or false if the timeout expired.
     */
    public boolean waitForPending(ResultCallback owner, long timeoutMs) {
        OwnerState state = activeOwners.get(owner);
        if (state != null) {
            try {
                return state.waitPending(timeoutMs);
            } catch (InterruptedException e) {
                return false;
            }
        }

        return true;
    }

    private OwnerState getNonNullState(ResultCallback owner) {
        OwnerState state = activeOwners.get(owner);
        if (state == null) {
            throw new IllegalStateException("No active callback : " + owner.toString());
        }
        return state;
    }

    public int getPendingCount(ResultCallback owner) {
        OwnerState state = activeOwners.get(owner);
        return (state != null) ? state.getNumPending() : 0;
    }<commit_msg>Remove unused code.<commit_after>// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.feedapi;

import com.yahoo.concurrent.SystemTimer;
import com.yahoo.jdisc.Metric;
import com.yahoo.log.LogLevel;
import com.yahoo.messagebus.*;
import com.yahoo.clientmetrics.RouteMetricSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

/**
 * This class allows multiple clients to use one shared messagebus session.
 * The user should create a ResultCallback, which acts as a "session" for that
 * client, and send one or more messages using the send() methods.
 * When done sending messages, the client can wait for all messages to be replied to
 * using the waitForPending method.
 */
public class SharedSender implements ReplyHandler {

    public static final Logger log = Logger.getLogger(SharedSender.class.getName());

    private SendSession sender;
    private RouteMetricSet metrics;
    private static final int REACT_LATENCY_ON_RACE = 5;
    private static final int REACT_LATENCY_ON_WATERMARK = 5;

    private ConcurrentHashMap<ResultCallback, OwnerState> activeOwners = new ConcurrentHashMap<>();

    /**
     * Creates a new shared sender.
     * If oldsender != null, we copy that status information from that sender.
     */
    public SharedSender(String route, SessionFactory factory, SharedSender oldSender, Metric metric) {
        if (factory != null) {
            sender = factory.createSendSession(this, metric);
        }

        if (oldSender != null) {
            this.metrics = oldSender.metrics;
        } else {
            metrics = new RouteMetricSet(route, null);
        }
    }

    public RouteMetricSet getMetrics() {
        return metrics;
    }

    public void remove(ResultCallback owner) {
        OwnerState state = activeOwners.remove(owner);
        if (state != null) {
            state.clearPending();
        }
    }

    public void shutdown() {
        try {
            while ( ! activeOwners.isEmpty()) {
                Thread.sleep(10);
            }
        } catch (InterruptedException e) {
        }
        sender.close();
    }

    /**
     * Waits until there are no more pending documents
     * for the given callback, or the timeout expires.
     *
     * @param owner     The callback to check for.
     * @param timeoutMs The number of milliseconds to wait, or -1 to wait indefinitely.
     * @return true if there were no more pending, or false if the timeout expired.
     */
    public boolean waitForPending(ResultCallback owner, long timeoutMs) {
        OwnerState state = activeOwners.get(owner);
        if (state != null) {
            try {
                return state.waitPending(timeoutMs);
            } catch (InterruptedException e) {
                return false;
            }
        }

        return true;
    }

    public int getPendingCount(ResultCallback owner) {
        OwnerState state = activeOwners.get(owner);
        return (state != null) ? state.getNumPending() : 0;
    }
{'fraction_non_alphanumeric': 0.08318014705882353, 'fraction_numerical': 0.003676470588235294, 'mean_word_length': 2.5561002178649237, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '829207', 'n_tokens_mistral': 1754, 'n_tokens_neox': 1680, 'n_words': 578}