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
151
152
153
154
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.search.result;
import com.yahoo.collections.ListenableArrayList;
import com.yahoo.component.provider.ListenableFreezableClass;
import com.yahoo.processing.Request;
import com.yahoo.processing.response.Data;
import com.yahoo.processing.response.DataList;
import com.yahoo.processing.response.DefaultIncomingData;
import com.yahoo.processing.response.IncomingData;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
/**
* A stream of events which can be rendered as Server-Sent Events (SSE).
*
* @author lesters
*/
public class EventStream extends Hit implements DataList<Data> {
private final ListenableArrayList<Data> data = new ListenableArrayList<>(16);
private final IncomingData<Data> incomingData;
private final AtomicInteger eventCount = new AtomicInteger(0);
public final static String EVENT_TYPE_TOKEN = "token";
public final static String DEFAULT_EVENT_TYPE = EVENT_TYPE_TOKEN;
public EventStream() {
super();
this.incomingData = new DefaultIncomingData<>(this);
}
public void add(String data) {
incoming().add(new Event(eventCount.incrementAndGet(), data, DEFAULT_EVENT_TYPE));
}
public void add(String data, String type) {
incoming().add(new Event(eventCount.incrementAndGet(), data, type));
}
public void error(String source, ErrorMessage message) {
incoming().add(new ErrorEvent(eventCount.incrementAndGet(), source, message));
}
public void markComplete() {
incoming().markComplete();
}
@Override
public Data add(Data event) {
data.add(event);
return event;
}
@Override
public Data get(int index) {
return data.get(index);
}
@Override
public List<Data> asList() {
return data;
}
@Override
public IncomingData<Data> incoming() {
return incomingData;
}
@Override
public CompletableFuture<DataList<Data>> completeFuture() {
return incomingData.completedFuture();
}
@Override
public void addDataListener(Runnable runnable) {
data.addListener(runnable);
}
@Override
public void close() {
}
public static class Event extends ListenableFreezableClass implements Data {
private final int eventNumber;
private final String data;
private final String type;
public Event(int eventNumber, String data, String type) {
this.eventNumber = eventNumber;
this.data = data;
this.type = type;
}
public String toString() {
return data;
}
public String type() {
return type;
}
@Override
public Request request() {
return null;
}
// For json rendering
public Hit asHit() {
Hit hit = new Hit(String.valueOf(eventNumber));
hit.setField(type, data);
return hit;
}
}
public static class ErrorEvent extends Event {
private final String source;
private final ErrorMessage message;
public ErrorEvent(int eventNumber, String source, ErrorMessage message) {
super(eventNumber, message.getMessage(), "error");
this.source = source;
this.message = message;
}
public String source() {
return source;
}
public int code() {
return message.getCode();
}
public String message() {
return message.getMessage();
}
@Override
public Hit asHit() {
Hit hit = super.asHit();
hit.setField("source", source);
hit.setField("code", message.getCode());
return hit;
}
}
}
|
