#5879

Committed 17 Dec 2025 04:15PM UTC coverage: 12.895% (-0.008%) from 12.903%

Build # #5879

Build Type

Pull #1641

github

Committed by

web-flow

Commit Message

Merge 9a2ba2654 into e14e97bb7

Pull Request Pull Request #1641: fix(boxsdkgen): replace internal links with absolute links (box/box-openapi#570)

Run Details

8368 of 64891 relevant lines covered (12.9%)

0.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/main/java/com/box/sdkgen/box/eventstream/EventStreamIterator.java

package com.box.sdkgen.box.eventstream;

import com.box.sdkgen.box.errors.BoxSDKError;
import com.box.sdkgen.managers.events.EventsManager;
import com.box.sdkgen.managers.events.GetEventStreamHeaders;
import com.box.sdkgen.managers.events.GetEventStreamQueryParams;
import com.box.sdkgen.managers.events.GetEventsHeaders;
import com.box.sdkgen.managers.events.GetEventsQueryParams;
import com.box.sdkgen.managers.events.GetEventsQueryParamsEventTypeField;
import com.box.sdkgen.managers.events.GetEventsQueryParamsStreamTypeField;
import com.box.sdkgen.networking.fetchoptions.FetchOptions;
import com.box.sdkgen.networking.fetchoptions.ResponseFormat;
import com.box.sdkgen.networking.fetchresponse.FetchResponse;
import com.box.sdkgen.schemas.event.Event;
import com.box.sdkgen.schemas.events.Events;
import com.box.sdkgen.schemas.realtimeserver.RealtimeServer;
import com.box.sdkgen.schemas.realtimeservers.RealtimeServers;
import com.box.sdkgen.serialization.json.Valuable;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

enum RealtimeServerEvent {
  NEW_CHANGE("new_change"),
  RECONNECT("reconnect");

  private final String value;

  RealtimeServerEvent(String value) {
    this.value = value;
  }

  public String getValue() {
    return value;
  }
}

enum EventStreamAction {
  FETCH_EVENTS,
  RECONNECT,
  RETRY,
  STOP
}

class EventStreamIterator implements Iterator<Event> {
  private static final int DEDUPLICATION_SIZE = 1000;
  private final EventsManager eventsManager;
  private final GetEventStreamQueryParams queryParams;
  private final GetEventStreamHeaders headersInput;
  private String streamPosition;
  private RealtimeServer longPollInfo;
  private volatile boolean stopped;
  private final CountDownLatch stopLatch;
  private final Map<String, Boolean> dedupHash;
  private final List<Event> eventQueue;
  private boolean started;
  private int longPollingRetries;
  private static final int DEFAULT_MAX_RETRIES = 10;

  EventStreamIterator(
      EventsManager eventsManager,
      GetEventStreamQueryParams queryParams,
      GetEventStreamHeaders headersInput) {
    this.eventsManager = eventsManager;
    this.queryParams = queryParams;
    this.headersInput = headersInput;
    this.streamPosition =
        queryParams.getStreamPosition() != null ? queryParams.getStreamPosition() : "now";
    this.longPollInfo = null;
    this.started = false;
    this.stopped = false;
    this.stopLatch = new CountDownLatch(1);
    this.dedupHash = new HashMap<>();
    this.eventQueue = new ArrayList<>();
    this.longPollingRetries = 0;
  }

  @Override
  public boolean hasNext() {
    return !eventQueue.isEmpty() || (!stopped && stopLatch.getCount() > 0);
  }

  @Override
  public Event next() {
    if (!hasNext()) {
      stop();
      throw new NoSuchElementException("No more events available or stream stopped");
    }
    if (!eventQueue.isEmpty()) {
      return eventQueue.remove(0);
    }
    if (!started) {
      started = true;
      fetchEvents();
      if (!eventQueue.isEmpty()) {
        return eventQueue.remove(0);
      }
    }

    getLongPollInfo();
    while (!stopped && stopLatch.getCount() > 0) {
      try {
        EventStreamAction action = doLongPoll();
        switch (action) {
          case FETCH_EVENTS:
            fetchEvents();
            if (!eventQueue.isEmpty()) {
              return eventQueue.remove(0);
            }
            break;
          case RETRY:
            try {
              Thread.sleep(5000);
            } catch (InterruptedException ie) {
              Thread.currentThread().interrupt();
            }
            continue;
          case RECONNECT:
            getLongPollInfo();
            continue;
          case STOP:
            stop();
            break;
        }
      } catch (Exception e) {
        boolean isBoxError = e instanceof BoxSDKError;
        if (!isBoxError) {
          throw e;
        }
      }
    }
    throw new NoSuchElementException("Stream stopped");
  }

  public void stop() {
    if (!stopped) {
      this.stopped = true;
      this.stopLatch.countDown();
    }
  }

  private void getLongPollInfo() throws BoxSDKError {
    if (stopped || stopLatch.getCount() == 0) {
      return;
    }

    try {
      RealtimeServers info = eventsManager.getEventsWithLongPolling();
      RealtimeServer server = null;
      if (info.getEntries() != null) {
        for (RealtimeServer entry : info.getEntries()) {
          if (entry.getType().equals("realtime_server")) {
            server = entry;
            break;
          }
        }
      }

      if (server == null) {
        throw new BoxSDKError("No realtime server found in the response.");
      }

      this.longPollInfo = server;
      this.longPollingRetries = 0;

    } catch (Exception e) {
      if (!stopped && stopLatch.getCount() > 0) {
        throw new BoxSDKError(
            "Failed to fetch long polling info: " + (e.getMessage() != null ? e.getMessage() : ""),
            e);
      }
    }
  }

  private EventStreamAction doLongPoll() throws BoxSDKError {
    if (stopped || stopLatch.getCount() == 0) {
      return EventStreamAction.STOP;
    }

    try {
      int maxRetries =
          (longPollInfo != null && longPollInfo.getMaxRetries() != null)
              ? Integer.parseInt(longPollInfo.getMaxRetries())
              : DEFAULT_MAX_RETRIES;

      if (longPollInfo == null || longPollingRetries > maxRetries) {
        getLongPollInfo();
      }

      longPollingRetries++;

      String longPollUrl = longPollInfo.getUrl();
      String separator = longPollUrl.contains("?") ? "&" : "?";
      String longPollWithStreamPosition =
          longPollUrl + separator + "stream_position=" + streamPosition;
      Map<String, String> headers = new HashMap<>();
      headers.put("Content-Type", "application/json");

      FetchResponse response =
          eventsManager
              .getNetworkSession()
              .getNetworkClient()
              .fetch(
                  new FetchOptions.Builder(longPollWithStreamPosition, "GET")
                      .responseFormat(ResponseFormat.JSON)
                      .auth(eventsManager.getAuth())
                      .networkSession(eventsManager.getNetworkSession())
                      .headers(headers)
                      .build());

      if (stopped || stopLatch.getCount() == 0) {
        return EventStreamAction.STOP;
      }

      if (response.getStatus() == 200 && response.getData() != null) {
        JsonNode message = response.getData();
        String messageText = message.has("message") ? message.get("message").asText() : null;

        if (RealtimeServerEvent.NEW_CHANGE.getValue().equals(messageText)) {
          return EventStreamAction.FETCH_EVENTS;
        } else if (RealtimeServerEvent.RECONNECT.getValue().equals(messageText)) {
          return EventStreamAction.RECONNECT;
        }
      }
      return EventStreamAction.RETRY;
    } catch (Exception e) {
      if (!stopped && stopLatch.getCount() > 0) {
        return EventStreamAction.RETRY;
      }
      return EventStreamAction.STOP;
    }
  }

  private void fetchEvents() throws BoxSDKError {
    if (stopped || stopLatch.getCount() == 0) {
      return;
    }

    try {
      GetEventsQueryParams.Builder fetchParamsBuilder =
          new GetEventsQueryParams.Builder().streamPosition(streamPosition);

      if (queryParams.getLimit() != null) {
        fetchParamsBuilder.limit(queryParams.getLimit());
      }
      if (queryParams.getCreatedAfter() != null) {
        fetchParamsBuilder.createdAfter(queryParams.getCreatedAfter());
      }
      if (queryParams.getCreatedBefore() != null) {
        fetchParamsBuilder.createdBefore(queryParams.getCreatedBefore());
      }

      if (queryParams.getStreamType() != null) {
        String streamTypeValue = queryParams.getStreamType().getValue();
        GetEventsQueryParamsStreamTypeField streamType = convertStreamType(streamTypeValue);
        fetchParamsBuilder.streamType(streamType);
      }

      List<? extends Valuable> eventTypes =
          queryParams.getEventType() != null
              ? queryParams.getEventType().stream()
                  .map(enumWrapper -> convertEventType(enumWrapper.getValue()))
                  .collect(Collectors.toList())
              : null;
      if (eventTypes != null) {
        fetchParamsBuilder.eventType(eventTypes);
      }

      GetEventsQueryParams fetchParams = fetchParamsBuilder.build();
      GetEventsHeaders fetchHeaders =
          new GetEventsHeaders.Builder()
              .extraHeaders(
                  headersInput.getExtraHeaders() != null
                      ? headersInput.getExtraHeaders()
                      : new HashMap<>())
              .build();

      Events events = eventsManager.getEvents(fetchParams, fetchHeaders);
      if (events.getEntries() != null && !events.getEntries().isEmpty()) {
        for (Event event : events.getEntries()) {
          String eventId = event.getEventId();
          if (eventId != null && !dedupHash.containsKey(eventId)) {
            dedupHash.put(eventId, true);
            if (stopped || stopLatch.getCount() == 0) {
              return;
            }
            eventQueue.add(event);
          }
        }
      }

      streamPosition =
          events.getNextStreamPosition() != null
              ? (events.getNextStreamPosition().isString()
                  ? events.getNextStreamPosition().getString()
                  : String.valueOf(events.getNextStreamPosition().getLongNumber()))
              : "now";

      if (dedupHash.size() >= DEDUPLICATION_SIZE) {
        dedupHash.clear();
        for (Event event : events.getEntries()) {
          String eventId = event.getEventId();
          if (eventId != null) {
            dedupHash.put(eventId, true);
          }
        }
      }
    } catch (Exception e) {
      if (!stopped && stopLatch.getCount() > 0) {
        if (e instanceof BoxSDKError) {
          throw (BoxSDKError) e;
        } else {
          throw new BoxSDKError(
              "Failed to fetch events: " + (e.getMessage() != null ? e.getMessage() : ""), e);
        }
      }
    }
  }

  private GetEventsQueryParamsStreamTypeField convertStreamType(String streamTypeValue) {
    if (streamTypeValue == null) return null;
    try {
      return GetEventsQueryParamsStreamTypeField.valueOf(
          streamTypeValue.toUpperCase().replace(" ", "_"));
    } catch (IllegalArgumentException e) {
      throw new BoxSDKError("Invalid stream type: " + streamTypeValue);
    }
  }

  private Valuable convertEventType(String eventTypeValue) {
    if (eventTypeValue == null) return null;
    try {
      return GetEventsQueryParamsEventTypeField.valueOf(
          eventTypeValue.toUpperCase().replace(" ", "_"));
    } catch (IllegalArgumentException e) {
      throw new BoxSDKError("Invalid event type: " + eventTypeValue);
    }
  }
}

1	package com.box.sdkgen.box.eventstream;
2
3	import com.box.sdkgen.box.errors.BoxSDKError;
4	import com.box.sdkgen.managers.events.EventsManager;
5	import com.box.sdkgen.managers.events.GetEventStreamHeaders;
6	import com.box.sdkgen.managers.events.GetEventStreamQueryParams;
7	import com.box.sdkgen.managers.events.GetEventsHeaders;
8	import com.box.sdkgen.managers.events.GetEventsQueryParams;
9	import com.box.sdkgen.managers.events.GetEventsQueryParamsEventTypeField;
10	import com.box.sdkgen.managers.events.GetEventsQueryParamsStreamTypeField;
11	import com.box.sdkgen.networking.fetchoptions.FetchOptions;
12	import com.box.sdkgen.networking.fetchoptions.ResponseFormat;
13	import com.box.sdkgen.networking.fetchresponse.FetchResponse;
14	import com.box.sdkgen.schemas.event.Event;
15	import com.box.sdkgen.schemas.events.Events;
16	import com.box.sdkgen.schemas.realtimeserver.RealtimeServer;
17	import com.box.sdkgen.schemas.realtimeservers.RealtimeServers;
18	import com.box.sdkgen.serialization.json.Valuable;
19	import com.fasterxml.jackson.databind.JsonNode;
20	import java.util.ArrayList;
21	import java.util.HashMap;
22	import java.util.Iterator;
23	import java.util.List;
24	import java.util.Map;
25	import java.util.NoSuchElementException;
26	import java.util.concurrent.CountDownLatch;
27	import java.util.stream.Collectors;
28
29	enum RealtimeServerEvent {	×
30	NEW_CHANGE("new_change"),	×
31	RECONNECT("reconnect");	×
32
33	private final String value;
34
35	RealtimeServerEvent(String value) {	×
36	this.value = value;	×
37	}	×
38
39	public String getValue() {
40	return value;	×
41	}
42	}
43
44	enum EventStreamAction {	×
45	FETCH_EVENTS,	×
46	RECONNECT,	×
47	RETRY,	×
48	STOP	×
49	}
50
51	class EventStreamIterator implements Iterator<Event> {
52	private static final int DEDUPLICATION_SIZE = 1000;
53	private final EventsManager eventsManager;
54	private final GetEventStreamQueryParams queryParams;
55	private final GetEventStreamHeaders headersInput;
56	private String streamPosition;
57	private RealtimeServer longPollInfo;
58	private volatile boolean stopped;
59	private final CountDownLatch stopLatch;
60	private final Map<String, Boolean> dedupHash;
61	private final List<Event> eventQueue;
62	private boolean started;
63	private int longPollingRetries;
64	private static final int DEFAULT_MAX_RETRIES = 10;
65
66	EventStreamIterator(
67	EventsManager eventsManager,
68	GetEventStreamQueryParams queryParams,
69	GetEventStreamHeaders headersInput) {	×
70	this.eventsManager = eventsManager;	×
71	this.queryParams = queryParams;	×
72	this.headersInput = headersInput;	×
73	this.streamPosition =	×
74	queryParams.getStreamPosition() != null ? queryParams.getStreamPosition() : "now";	×
75	this.longPollInfo = null;	×
76	this.started = false;	×
77	this.stopped = false;	×
78	this.stopLatch = new CountDownLatch(1);	×
79	this.dedupHash = new HashMap<>();	×
80	this.eventQueue = new ArrayList<>();	×
81	this.longPollingRetries = 0;	×
82	}	×
83
84	@Override
85	public boolean hasNext() {
86	return !eventQueue.isEmpty() \|\| (!stopped && stopLatch.getCount() > 0);	×
87	}
88
89	@Override
90	public Event next() {
91	if (!hasNext()) {	×
92	stop();	×
93	throw new NoSuchElementException("No more events available or stream stopped");	×
94	}
95	if (!eventQueue.isEmpty()) {	×
96	return eventQueue.remove(0);	×
97	}
98	if (!started) {	×
99	started = true;	×
100	fetchEvents();	×
101	if (!eventQueue.isEmpty()) {	×
102	return eventQueue.remove(0);	×
103	}
104	}
105
106	getLongPollInfo();	×
107	while (!stopped && stopLatch.getCount() > 0) {	×
108	try {
109	EventStreamAction action = doLongPoll();	×
110	switch (action) {	×
111	case FETCH_EVENTS:
112	fetchEvents();	×
113	if (!eventQueue.isEmpty()) {	×
114	return eventQueue.remove(0);	×
115	}
116	break;
117	case RETRY:
118	try {
119	Thread.sleep(5000);	×
120	} catch (InterruptedException ie) {	×
121	Thread.currentThread().interrupt();	×
122	}	×
123	continue;	×
124	case RECONNECT:
125	getLongPollInfo();	×
126	continue;	×
127	case STOP:
128	stop();	×
129	break;
130	}
131	} catch (Exception e) {	×
132	boolean isBoxError = e instanceof BoxSDKError;	×
133	if (!isBoxError) {	×
134	throw e;	×
135	}
136	}	×
137	}
138	throw new NoSuchElementException("Stream stopped");	×
139	}
140
141	public void stop() {
142	if (!stopped) {	×
143	this.stopped = true;	×
144	this.stopLatch.countDown();	×
145	}
146	}	×
147
148	private void getLongPollInfo() throws BoxSDKError {
149	if (stopped \|\| stopLatch.getCount() == 0) {	×
150	return;	×
151	}
152
153	try {
154	RealtimeServers info = eventsManager.getEventsWithLongPolling();	×
155	RealtimeServer server = null;	×
156	if (info.getEntries() != null) {	×
157	for (RealtimeServer entry : info.getEntries()) {	×
158	if (entry.getType().equals("realtime_server")) {	×
159	server = entry;	×
160	break;	×
161	}
162	}	×
163	}
164
165	if (server == null) {	×
166	throw new BoxSDKError("No realtime server found in the response.");	×
167	}
168
169	this.longPollInfo = server;	×
170	this.longPollingRetries = 0;	×
171
172	} catch (Exception e) {	×
173	if (!stopped && stopLatch.getCount() > 0) {	×
174	throw new BoxSDKError(	×
175	"Failed to fetch long polling info: " + (e.getMessage() != null ? e.getMessage() : ""),	×
176	e);
177	}
178	}	×
179	}	×
180
181	private EventStreamAction doLongPoll() throws BoxSDKError {
182	if (stopped \|\| stopLatch.getCount() == 0) {	×
183	return EventStreamAction.STOP;	×
184	}
185
186	try {
187	int maxRetries =	×
188	(longPollInfo != null && longPollInfo.getMaxRetries() != null)	×
189	? Integer.parseInt(longPollInfo.getMaxRetries())	×
190	: DEFAULT_MAX_RETRIES;
191
192	if (longPollInfo == null \|\| longPollingRetries > maxRetries) {	×
193	getLongPollInfo();	×
194	}
195
196	longPollingRetries++;	×
197
198	String longPollUrl = longPollInfo.getUrl();	×
199	String separator = longPollUrl.contains("?") ? "&" : "?";	×
200	String longPollWithStreamPosition =	×
201	longPollUrl + separator + "stream_position=" + streamPosition;
202	Map<String, String> headers = new HashMap<>();	×
203	headers.put("Content-Type", "application/json");	×
204
205	FetchResponse response =	×
206	eventsManager
207	.getNetworkSession()	×
208	.getNetworkClient()	×
209	.fetch(	×
210	new FetchOptions.Builder(longPollWithStreamPosition, "GET")
211	.responseFormat(ResponseFormat.JSON)	×
212	.auth(eventsManager.getAuth())	×
213	.networkSession(eventsManager.getNetworkSession())	×
214	.headers(headers)	×
215	.build());	×
216
217	if (stopped \|\| stopLatch.getCount() == 0) {	×
218	return EventStreamAction.STOP;	×
219	}
220
221	if (response.getStatus() == 200 && response.getData() != null) {	×
222	JsonNode message = response.getData();	×
223	String messageText = message.has("message") ? message.get("message").asText() : null;	×
224
225	if (RealtimeServerEvent.NEW_CHANGE.getValue().equals(messageText)) {	×
226	return EventStreamAction.FETCH_EVENTS;	×
227	} else if (RealtimeServerEvent.RECONNECT.getValue().equals(messageText)) {	×
228	return EventStreamAction.RECONNECT;	×
229	}
230	}
231	return EventStreamAction.RETRY;	×
232	} catch (Exception e) {	×
233	if (!stopped && stopLatch.getCount() > 0) {	×
234	return EventStreamAction.RETRY;	×
235	}
236	return EventStreamAction.STOP;	×
237	}
238	}
239
240	private void fetchEvents() throws BoxSDKError {
241	if (stopped \|\| stopLatch.getCount() == 0) {	×
242	return;	×
243	}
244
245	try {
246	GetEventsQueryParams.Builder fetchParamsBuilder =	×
247	new GetEventsQueryParams.Builder().streamPosition(streamPosition);	×
248
249	if (queryParams.getLimit() != null) {	×
250	fetchParamsBuilder.limit(queryParams.getLimit());	×
251	}
252	if (queryParams.getCreatedAfter() != null) {	×
253	fetchParamsBuilder.createdAfter(queryParams.getCreatedAfter());	×
254	}
255	if (queryParams.getCreatedBefore() != null) {	×
256	fetchParamsBuilder.createdBefore(queryParams.getCreatedBefore());	×
257	}
258
259	if (queryParams.getStreamType() != null) {	×
260	String streamTypeValue = queryParams.getStreamType().getValue();	×
261	GetEventsQueryParamsStreamTypeField streamType = convertStreamType(streamTypeValue);	×
262	fetchParamsBuilder.streamType(streamType);	×
263	}
264
265	List<? extends Valuable> eventTypes =	×
266	queryParams.getEventType() != null	×
267	? queryParams.getEventType().stream()	×
268	.map(enumWrapper -> convertEventType(enumWrapper.getValue()))	×
269	.collect(Collectors.toList())	×
270	: null;
271	if (eventTypes != null) {	×
272	fetchParamsBuilder.eventType(eventTypes);	×
273	}
274
275	GetEventsQueryParams fetchParams = fetchParamsBuilder.build();	×
276	GetEventsHeaders fetchHeaders =	×
277	new GetEventsHeaders.Builder()
278	.extraHeaders(	×
279	headersInput.getExtraHeaders() != null	×
280	? headersInput.getExtraHeaders()	×
281	: new HashMap<>())
282	.build();	×
283
284	Events events = eventsManager.getEvents(fetchParams, fetchHeaders);	×
285	if (events.getEntries() != null && !events.getEntries().isEmpty()) {	×
286	for (Event event : events.getEntries()) {	×
287	String eventId = event.getEventId();	×
288	if (eventId != null && !dedupHash.containsKey(eventId)) {	×
289	dedupHash.put(eventId, true);	×
290	if (stopped \|\| stopLatch.getCount() == 0) {	×
291	return;	×
292	}
293	eventQueue.add(event);	×
294	}
295	}	×
296	}
297
298	streamPosition =	×
299	events.getNextStreamPosition() != null	×
300	? (events.getNextStreamPosition().isString()	×
301	? events.getNextStreamPosition().getString()	×
302	: String.valueOf(events.getNextStreamPosition().getLongNumber()))	×
303	: "now";
304
305	if (dedupHash.size() >= DEDUPLICATION_SIZE) {	×
306	dedupHash.clear();	×
307	for (Event event : events.getEntries()) {	×
308	String eventId = event.getEventId();	×
309	if (eventId != null) {	×
310	dedupHash.put(eventId, true);	×
311	}
312	}	×
313	}
314	} catch (Exception e) {	×
315	if (!stopped && stopLatch.getCount() > 0) {	×
316	if (e instanceof BoxSDKError) {	×
317	throw (BoxSDKError) e;	×
318	} else {
319	throw new BoxSDKError(	×
320	"Failed to fetch events: " + (e.getMessage() != null ? e.getMessage() : ""), e);	×
321	}
322	}
323	}	×
324	}	×
325
326	private GetEventsQueryParamsStreamTypeField convertStreamType(String streamTypeValue) {
327	if (streamTypeValue == null) return null;	×
328	try {
329	return GetEventsQueryParamsStreamTypeField.valueOf(	×
330	streamTypeValue.toUpperCase().replace(" ", "_"));	×
331	} catch (IllegalArgumentException e) {	×
332	throw new BoxSDKError("Invalid stream type: " + streamTypeValue);	×
333	}
334	}
335
336	private Valuable convertEventType(String eventTypeValue) {
337	if (eventTypeValue == null) return null;	×
338	try {
339	return GetEventsQueryParamsEventTypeField.valueOf(	×
340	eventTypeValue.toUpperCase().replace(" ", "_"));	×
341	} catch (IllegalArgumentException e) {	×
342	throw new BoxSDKError("Invalid event type: " + eventTypeValue);	×
343	}
344	}
345	}

box / box-java-sdk / #5879

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous