blob: 4a85f582905a4d6817eb9409bb8b1eab976223f4 (
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
// Copyright 2016 Yahoo Inc. 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.ExecutionList;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import com.yahoo.collections.Tuple2;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
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 SettableFuture<DataList<DATATYPE>> completionFuture;
private final List<DATATYPE> dataList = new ArrayList<>();
private List<Tuple2<Runnable,Executor>> newDataListeners = null;
/**
* If this is completed 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 = SettableFuture.create();
}
/**
* 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
public ListenableFuture<DataList<DATATYPE>> completed() {
return completionFuture;
}
/**
* Returns whether the data in this is complete
*/
@Override
public synchronized boolean isComplete() {
return complete;
}
/**
* Add new data and mark this as completed
*/
@Override
public synchronized void addLast(DATATYPE data) {
addLast(Collections.singletonList(data));
}
/**
* Add new data without completing this
*/
@Override
public synchronized void add(DATATYPE data) {
add(Collections.singletonList(data));
}
/**
* Add new data and mark this as completed
*/
@Override
public synchronized void addLast(List<DATATYPE> data) {
add(data);
markComplete();
}
/**
* Add 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();
}
/**
* Mark this as completed and notify any listeners
*/
@Override
public synchronized void markComplete() {
complete = true;
completionFuture.set(owner);
}
/**
* Get and remove 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;
}
}
|
