OutboxImpl implements OutboxInternal { @Override public final boolean offerToSnapshot(@Nonnull Object key, @Nonnull Object value) { if (snapshotEdge == null) { throw new IllegalStateException("Outbox does not have snapshot queue"); } if (shouldBlock()) { return false; } assert unfinishedSnapshotKey == null || unfinishedSnapshotKey.equals(key) : "Different key offered after previous call returned false: expected=" + unfinishedSnapshotKey + ", got=" + key; assert unfinishedSnapshotValue == null || unfinishedSnapshotValue.equals(value) : "Different value offered after previous call returned false: expected=" + unfinishedSnapshotValue + ", got=" + value; if (pendingSnapshotEntry == null) { Data sKey = serializationService.toData(key); Data sValue = serializationService.toData(value); pendingSnapshotEntry = entry(sKey, sValue); } boolean success = offerInternal(snapshotEdge, pendingSnapshotEntry); if (success) { pendingSnapshotEntry = null; unfinishedSnapshotKey = null; unfinishedSnapshotValue = null; } else { unfinishedSnapshotKey = key; unfinishedSnapshotValue = value; } return success; } OutboxImpl(OutboundCollector[] outstreams, boolean hasSnapshot, ProgressTracker progTracker,
                      SerializationService serializationService, int batchSize, AtomicLongArray counters); @Override final int bucketCount(); @Override final boolean offer(int ordinal, @Nonnull Object item); @Override final boolean offer(@Nonnull int[] ordinals, @Nonnull Object item); @Override final boolean offer(@Nonnull Object item); @Override final boolean offerToSnapshot(@Nonnull Object key, @Nonnull Object value); @Override boolean hasUnfinishedItem(); @Override void block(); @Override void unblock(); @Override void reset(); @Override long lastForwardedWm();  }
@Test public void when_offer3_then_rateLimited() { do_when_offer_then_rateLimited(e -> outbox.offerToSnapshot(e, e)); }
@Test public void when_offer3FailsAndDifferentItemOffered_then_fail() { do_when_offerDifferent_then_fail(e -> outbox.offerToSnapshot(e, e)); }