vespa-http-client/src/main/java/com/yahoo/vespa/http/client/core/api/SessionImpl.java


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.http.client.core.api;

import com.yahoo.vespa.http.client.FeedClient;
import com.yahoo.vespa.http.client.Result;
import com.yahoo.vespa.http.client.config.SessionParams;
import com.yahoo.vespa.http.client.core.ThrottlePolicy;
import com.yahoo.vespa.http.client.core.operationProcessor.IncompleteResultsThrottler;
import com.yahoo.vespa.http.client.core.operationProcessor.OperationProcessor;

import java.io.OutputStream;
import java.time.Clock;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * This class wires up the Session API using MultiClusterHandler and MultiClusterSessionOutputStream.
 *
 * @deprecated
 */
@Deprecated // TODO: Remove on Vespa 8
public class SessionImpl implements com.yahoo.vespa.http.client.Session {

    private final OperationProcessor operationProcessor;
    private final BlockingQueue<Result> resultQueue = new LinkedBlockingQueue<>();


    public SessionImpl(SessionParams sessionParams, ScheduledThreadPoolExecutor timeoutExecutor, Clock clock) {
        this.operationProcessor = new OperationProcessor(
                new IncompleteResultsThrottler(
                        sessionParams.getThrottlerMinSize(),
                        sessionParams.getClientQueueSize(),
                        ()->System.currentTimeMillis(),
                        new ThrottlePolicy()),
                new FeedClient.ResultCallback() {
                    @Override
                    public void onCompletion(String docId, Result documentResult) {
                        resultQueue.offer(documentResult);
                    }
                },
                sessionParams,
                timeoutExecutor,
                clock);
    }

    @Override
    public OutputStream stream(CharSequence documentId) {
        return new MultiClusterSessionOutputStream(documentId, operationProcessor, null);
    }

    @Override
    public BlockingQueue<Result> results() {
        return resultQueue;
    }

    @Override
    public void close() {
        operationProcessor.close();
    }

    @Override
    public String getStatsAsJson() {
        return operationProcessor.getStatsAsJson();
    }

    // For testing only (legacy tests).
    public int getIncompleteResultQueueSize() {
        return operationProcessor.getIncompleteResultQueueSize();
    }

}