<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.documentapi;

import com.yahoo.document.DocumentOperation;
import com.yahoo.documentapi.messagebus.protocol.PutDocumentMessage;
import com.yahoo.documentapi.messagebus.protocol.RemoveDocumentMessage;
import com.yahoo.messagebus.Message;
import com.yahoo.vdslib.DocumentList;
import com.yahoo.vdslib.Entry;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * A visitor data handler that queues up documents in visitor responses and
 * implements the <code>getNext</code> methods, thus implementing the polling
 * API defined in VisitorDataHandler.
 * <p>
 * Visitor responses containing document lists should be polled for with the
 * <code>getNext</code> methods and need to be acked when processed for
 * visiting not to halt. The class is thread safe.
 *
 * @author <NAME>
 * @author vekterli
 */
public class VisitorDataQueue extends VisitorDataHandler {

    final BlockingQueue<VisitorResponse> pendingResponses = new LinkedBlockingQueue<>();

    /** Creates a new visitor data queue. */
    public VisitorDataQueue() {
    }

    // Inherit doc from VisitorDataHandler
    @Override
    public void reset() {
        super.reset();
        pendingResponses.clear();
    }

    private void appendSingleOpToPendingList(final DocumentOperation op, final AckToken token) {
        final DocumentList docList = DocumentList.create(Entry.create(op));
        final DocumentListVisitorResponse response = new DocumentListVisitorResponse(docList, token);
        pendingResponses.add(response);
    }

    @Override
    public void onMessage(Message m, AckToken token) {
        if (m instanceof PutDocumentMessage) {
            appendSingleOpToPendingList(((PutDocumentMessage)m).getDocumentPut(), token);
        } else if (m instanceof RemoveDocumentMessage) {
            appendSingleOpToPendingList(((RemoveDocumentMessage)m).getDocumentRemove(), token);
        } else {
            throw new UnsupportedOperationException(
                    String.format("Expected put/remove message, got '%s' of type %s",<commit_msg>Make member var for blocking queue private<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.documentapi;

import com.yahoo.document.DocumentOperation;
import com.yahoo.documentapi.messagebus.protocol.PutDocumentMessage;
import com.yahoo.documentapi.messagebus.protocol.RemoveDocumentMessage;
import com.yahoo.messagebus.Message;
import com.yahoo.vdslib.DocumentList;
import com.yahoo.vdslib.Entry;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * A visitor data handler that queues up documents in visitor responses and
 * implements the <code>getNext</code> methods, thus implementing the polling
 * API defined in VisitorDataHandler.
 * <p>
 * Visitor responses containing document lists should be polled for with the
 * <code>getNext</code> methods and need to be acked when processed for
 * visiting not to halt. The class is thread safe.
 *
 * @author <NAME>
 * @author vekterli
 */
public class VisitorDataQueue extends VisitorDataHandler {

    private final BlockingQueue<VisitorResponse> pendingResponses = new LinkedBlockingQueue<>();

    /** Creates a new visitor data queue. */
    public VisitorDataQueue() {
    }

    // Inherit doc from VisitorDataHandler
    @Override
    public void reset() {
        super.reset();
        pendingResponses.clear();
    }

    private void appendSingleOpToPendingList(final DocumentOperation op, final AckToken token) {
        final DocumentList docList = DocumentList.create(Entry.create(op));
        final DocumentListVisitorResponse response = new DocumentListVisitorResponse(docList, token);
        pendingResponses.add(response);
    }

    @Override
    public void onMessage(Message m, AckToken token) {
        if (m instanceof PutDocumentMessage) {
            appendSingleOpToPendingList(((PutDocumentMessage)m).getDocumentPut(), token);
        } else if (m instanceof RemoveDocumentMessage) {
            appendSingleOpToPendingList(((RemoveDocumentMessage)m).getDocumentRemove(), token);
        } else {
            throw new UnsupportedOperationException(
                    String.format("Expected put/remove message, got '%s' of type %s",
{'fraction_non_alphanumeric': 0.08371685294762218, 'fraction_numerical': 0.0026298487836949377, 'mean_word_length': 3.9608695652173913, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, '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': '4741280', 'n_tokens_mistral': 1204, 'n_tokens_neox': 1167, 'n_words': 362}