@Test public void testTaskRead() { TaskQueue queue = new InMemorySubmissionQueue<>(); UUID proxyId = UUID.randomUUID(); DataSubmissionTask<? extends DataSubmissionTask<?>> task = null; if (this.expectedTask instanceof LineDelimitedDataSubmissionTask) { task = new LineDelimitedDataSubmissionTask(null, proxyId, new DefaultEntityPropertiesForTesting(), queue, "wavefront", ReportableEntityType.POINT, "2878", ImmutableList.of("item1", "item2", "item3"), time::get); } else if (this.expectedTask instanceof EventDataSubmissionTask) { task = new EventDataSubmissionTask(null, proxyId, new DefaultEntityPropertiesForTesting(), queue, "2878", ImmutableList.of( new Event(ReportEvent.newBuilder(). setStartTime(time.get() * 1000). setEndTime(time.get() * 1000 + 1). setName("Event name for testing"). setHosts(ImmutableList.of("host1", "host2")). setDimensions(ImmutableMap.of("multi", ImmutableList.of("bar", "baz"))). setAnnotations(ImmutableMap.of("severity", "INFO")). setTags(ImmutableList.of("tag1")). build())), time::get); } else if (this.expectedTask instanceof SourceTagSubmissionTask) { task = new SourceTagSubmissionTask(null, new DefaultEntityPropertiesForTesting(), queue, "2878", new SourceTag( ReportSourceTag.newBuilder().setOperation(SourceOperationType.SOURCE_TAG). setAction(SourceTagAction.SAVE).setSource("testSource"). setAnnotations(ImmutableList.of("newtag1", "newtag2")).build()), time::get); } assertNotNull(task); task.enqueue(QueueingReason.RETRY); if (this.expectedTask instanceof LineDelimitedDataSubmissionTask) { LineDelimitedDataSubmissionTask readTask = (LineDelimitedDataSubmissionTask) queue.peek(); assertNotNull(readTask); assertEquals(((LineDelimitedDataSubmissionTask) task).payload(), readTask.payload()); assertEquals(((LineDelimitedDataSubmissionTask) this.expectedTask).payload(), readTask.payload()); assertEquals(77777, readTask.getEnqueuedMillis()); } if (this.expectedTask instanceof EventDataSubmissionTask) { EventDataSubmissionTask readTask = (EventDataSubmissionTask) queue.peek(); assertNotNull(readTask); assertEquals(((EventDataSubmissionTask) task).payload(), readTask.payload()); assertEquals(((EventDataSubmissionTask) this.expectedTask).payload(), readTask.payload()); assertEquals(77777, readTask.getEnqueuedMillis()); } if (this.expectedTask instanceof SourceTagSubmissionTask) { SourceTagSubmissionTask readTask = (SourceTagSubmissionTask) queue.peek(); assertNotNull(readTask); assertEquals(((SourceTagSubmissionTask) task).payload(), readTask.payload()); assertEquals(((SourceTagSubmissionTask) this.expectedTask).payload(), readTask.payload()); assertEquals(77777, readTask.getEnqueuedMillis()); } }
@Nullable @Override public T peek() { try { if (this.head != null) return this.head; this.head = wrapped.peek(); return this.head; } catch (IOException ex) { throw Utils.<Error>throwAny(ex); } }
InMemorySubmissionQueue implements TaskQueue<T> { @Nullable @Override public T peek() { try { if (this.head != null) return this.head; this.head = wrapped.peek(); return this.head; } catch (IOException ex) { throw Utils.<Error>throwAny(ex); } } }
InMemorySubmissionQueue implements TaskQueue<T> { @Nullable @Override public T peek() { try { if (this.head != null) return this.head; this.head = wrapped.peek(); return this.head; } catch (IOException ex) { throw Utils.<Error>throwAny(ex); } } InMemorySubmissionQueue(); }
InMemorySubmissionQueue implements TaskQueue<T> { @Nullable @Override public T peek() { try { if (this.head != null) return this.head; this.head = wrapped.peek(); return this.head; } catch (IOException ex) { throw Utils.<Error>throwAny(ex); } } InMemorySubmissionQueue(); @Override int size(); @Nullable @Override Long weight(); @Nullable @Override Long getAvailableBytes(); @Override String getName(); @Nullable @Override T peek(); @Override void add(@Nonnull T entry); @Override void clear(); @Override void remove(); @Override void close(); @NotNull @Override Iterator<T> iterator(); }
InMemorySubmissionQueue implements TaskQueue<T> { @Nullable @Override public T peek() { try { if (this.head != null) return this.head; this.head = wrapped.peek(); return this.head; } catch (IOException ex) { throw Utils.<Error>throwAny(ex); } } InMemorySubmissionQueue(); @Override int size(); @Nullable @Override Long weight(); @Nullable @Override Long getAvailableBytes(); @Override String getName(); @Nullable @Override T peek(); @Override void add(@Nonnull T entry); @Override void clear(); @Override void remove(); @Override void close(); @NotNull @Override Iterator<T> iterator();  }