@Test void testDomainEventStream_nullPointerExceptionOnEmptyEventStream() { DomainEventStream eventStream = EventStreamUtils.upcastAndDeserializeDomainEvents(Stream.empty(), serializer, NoOpEventUpcaster.INSTANCE); assertNull(eventStream.getLastSequenceNumber()); }
@SuppressWarnings("OptionalGetWithoutIsPresent") public static DomainEventStream upcastAndDeserializeDomainEvents( Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain) { AtomicReference<Long> currentSequenceNumber = new AtomicReference<>(); Stream<IntermediateEventRepresentation> upcastResult = upcastAndDeserialize(eventEntryStream, upcasterChain, entry -> { InitialEventRepresentation result = new InitialEventRepresentation(entry, serializer); currentSequenceNumber.set(result.getSequenceNumber().get()); return result; }); Stream<? extends DomainEventMessage<?>> stream = upcastResult.map(ir -> { SerializedMessage<?> serializedMessage = new SerializedMessage<>(ir.getMessageIdentifier(), new LazyDeserializingObject<>( ir::getData, ir.getType(), serializer), ir.getMetaData()); if (ir.getTrackingToken().isPresent()) { return new GenericTrackedDomainEventMessage<>(ir.getTrackingToken().get(), ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } else { return new GenericDomainEventMessage<>(ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } }); return DomainEventStream.of(stream, currentSequenceNumber::get); }
EventStreamUtils { @SuppressWarnings("OptionalGetWithoutIsPresent") public static DomainEventStream upcastAndDeserializeDomainEvents( Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain) { AtomicReference<Long> currentSequenceNumber = new AtomicReference<>(); Stream<IntermediateEventRepresentation> upcastResult = upcastAndDeserialize(eventEntryStream, upcasterChain, entry -> { InitialEventRepresentation result = new InitialEventRepresentation(entry, serializer); currentSequenceNumber.set(result.getSequenceNumber().get()); return result; }); Stream<? extends DomainEventMessage<?>> stream = upcastResult.map(ir -> { SerializedMessage<?> serializedMessage = new SerializedMessage<>(ir.getMessageIdentifier(), new LazyDeserializingObject<>( ir::getData, ir.getType(), serializer), ir.getMetaData()); if (ir.getTrackingToken().isPresent()) { return new GenericTrackedDomainEventMessage<>(ir.getTrackingToken().get(), ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } else { return new GenericDomainEventMessage<>(ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } }); return DomainEventStream.of(stream, currentSequenceNumber::get); } }
EventStreamUtils { @SuppressWarnings("OptionalGetWithoutIsPresent") public static DomainEventStream upcastAndDeserializeDomainEvents( Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain) { AtomicReference<Long> currentSequenceNumber = new AtomicReference<>(); Stream<IntermediateEventRepresentation> upcastResult = upcastAndDeserialize(eventEntryStream, upcasterChain, entry -> { InitialEventRepresentation result = new InitialEventRepresentation(entry, serializer); currentSequenceNumber.set(result.getSequenceNumber().get()); return result; }); Stream<? extends DomainEventMessage<?>> stream = upcastResult.map(ir -> { SerializedMessage<?> serializedMessage = new SerializedMessage<>(ir.getMessageIdentifier(), new LazyDeserializingObject<>( ir::getData, ir.getType(), serializer), ir.getMetaData()); if (ir.getTrackingToken().isPresent()) { return new GenericTrackedDomainEventMessage<>(ir.getTrackingToken().get(), ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } else { return new GenericDomainEventMessage<>(ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } }); return DomainEventStream.of(stream, currentSequenceNumber::get); } private  EventStreamUtils(); }
EventStreamUtils { @SuppressWarnings("OptionalGetWithoutIsPresent") public static DomainEventStream upcastAndDeserializeDomainEvents( Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain) { AtomicReference<Long> currentSequenceNumber = new AtomicReference<>(); Stream<IntermediateEventRepresentation> upcastResult = upcastAndDeserialize(eventEntryStream, upcasterChain, entry -> { InitialEventRepresentation result = new InitialEventRepresentation(entry, serializer); currentSequenceNumber.set(result.getSequenceNumber().get()); return result; }); Stream<? extends DomainEventMessage<?>> stream = upcastResult.map(ir -> { SerializedMessage<?> serializedMessage = new SerializedMessage<>(ir.getMessageIdentifier(), new LazyDeserializingObject<>( ir::getData, ir.getType(), serializer), ir.getMetaData()); if (ir.getTrackingToken().isPresent()) { return new GenericTrackedDomainEventMessage<>(ir.getTrackingToken().get(), ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } else { return new GenericDomainEventMessage<>(ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } }); return DomainEventStream.of(stream, currentSequenceNumber::get); } private  EventStreamUtils(); @SuppressWarnings("OptionalGetWithoutIsPresent") static DomainEventStream upcastAndDeserializeDomainEvents(
            Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain); static Stream<? extends DomainEventMessage<?>> asStream(DomainEventStream domainEventStream); }
EventStreamUtils { @SuppressWarnings("OptionalGetWithoutIsPresent") public static DomainEventStream upcastAndDeserializeDomainEvents( Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain) { AtomicReference<Long> currentSequenceNumber = new AtomicReference<>(); Stream<IntermediateEventRepresentation> upcastResult = upcastAndDeserialize(eventEntryStream, upcasterChain, entry -> { InitialEventRepresentation result = new InitialEventRepresentation(entry, serializer); currentSequenceNumber.set(result.getSequenceNumber().get()); return result; }); Stream<? extends DomainEventMessage<?>> stream = upcastResult.map(ir -> { SerializedMessage<?> serializedMessage = new SerializedMessage<>(ir.getMessageIdentifier(), new LazyDeserializingObject<>( ir::getData, ir.getType(), serializer), ir.getMetaData()); if (ir.getTrackingToken().isPresent()) { return new GenericTrackedDomainEventMessage<>(ir.getTrackingToken().get(), ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } else { return new GenericDomainEventMessage<>(ir.getAggregateType().get(), ir.getAggregateIdentifier().get(), ir.getSequenceNumber().get(), serializedMessage, ir::getTimestamp); } }); return DomainEventStream.of(stream, currentSequenceNumber::get); } private  EventStreamUtils(); @SuppressWarnings("OptionalGetWithoutIsPresent") static DomainEventStream upcastAndDeserializeDomainEvents(
            Stream<? extends DomainEventData<?>> eventEntryStream, Serializer serializer, EventUpcaster upcasterChain); static Stream<? extends DomainEventMessage<?>> asStream(DomainEventStream domainEventStream);  }