blob: 813d6ac54d8b3a84c4ad152e8edeebfc5aeae4e6 (
plain) (
blame)
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.processing.response;
import com.google.common.util.concurrent.ListenableFuture;
import com.yahoo.collections.Tuple2;
import com.yahoo.concurrent.CompletableFutures;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
/**
* The default incoming data implementation
*
* @author bratseth
*/
public class DefaultIncomingData<DATATYPE extends Data> implements IncomingData<DATATYPE> {
private DataList<DATATYPE> owner = null;
private final CompletableFuture<DataList<DATATYPE>> completionFuture;
private final List<DATATYPE> dataList = new ArrayList<>();
private List<Tuple2<Runnable,Executor>> newDataListeners = null;
/** Whether this is completed, such that no more data can be added */
private boolean complete = false;
/** Creates an instance which must be assigned an owner after creation */
public DefaultIncomingData() {
this(null);
}
public DefaultIncomingData(DataList<DATATYPE> owner) {
assignOwner(owner);
completionFuture = new CompletableFuture<>();
}
/** Assigns the owner of this. Throws an exception if the owner is already set. */
public final void assignOwner(DataList<DATATYPE> owner) {
if (this.owner != null) throw new NullPointerException("Owner of " + this + " was already assigned");
this.owner = owner;
}
@Override
public DataList<DATATYPE> getOwner() {
return owner;
}
@Override
@Deprecated(forRemoval = true, since = "7")
@SuppressWarnings("removal")
public ListenableFuture<DataList<DATATYPE>> completed() {
return CompletableFutures.toGuavaListenableFuture(completionFuture);
}
@Override public CompletableFuture<DataList<DATATYPE>> completedFuture() { return completionFuture; }
/** Returns whether the data in this is complete */
@Override
public synchronized boolean isComplete() {
return complete;
}
/** Adds new data and marks this as completed */
@Override
public synchronized void addLast(DATATYPE data) {
addLast(Collections.singletonList(data));
}
/** Adds new data without completing this */
@Override
public synchronized void add(DATATYPE data) {
add(Collections.singletonList(data));
}
/** Adds new data and marks this as completed */
@Override
public synchronized void addLast(List<DATATYPE> data) {
add(data);
markComplete();
}
/** Adds new data without completing this */
@Override
public synchronized void add(List<DATATYPE> data) {
if (complete) throw new IllegalStateException("Attempted to add data to completed " + this);
dataList.addAll(data);
notifyDataListeners();
}
/** Marks this as completed and notify any listeners */
@Override
public synchronized void markComplete() {
complete = true;
completionFuture.complete(owner);
}
/**
* Gets and removes all the data currently available in this.
* The returned list is a modifiable fresh instance owned by the caller.
*/
public synchronized List<DATATYPE> drain() {
List<DATATYPE> dataListCopy = new ArrayList<>(dataList);
dataList.clear();
return dataListCopy;
}
@Override
public void addNewDataListener(Runnable listener, Executor executor) {
synchronized (this) {
if (newDataListeners == null)
newDataListeners = new ArrayList<>();
newDataListeners.add(new Tuple2<>(listener, executor));
if (dataList.isEmpty()) return;
}
notifyDataListeners();
}
private void notifyDataListeners() {
if (newDataListeners == null) return;
for (Tuple2<Runnable, Executor> listener : newDataListeners) {
listener.second.execute(listener.first);
}
}
@Override
public String toString() {
return "incoming: " + (complete ? "complete" : "incomplete") + ", data " + dataList;
}
}
|
