java.util.ArrayDeque#offer ( )源码实例Demo

static void arrayDequeStuff() {
  ArrayDeque<Object> d = new ArrayDeque<>();
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.add(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.addFirst(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.addLast(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.offerFirst(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.offerLast(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.offer(null);
  // BUG: Diagnostic contains: passing @Nullable parameter 'null' where @NonNull is required
  d.push(null);
  Object[] o = null;
  // BUG: Diagnostic contains: passing @Nullable parameter 'o' where @NonNull is required
  d.toArray(o);
  // this should be fine
  d.toArray();
}
 

@Test
public void postCompleteWithRequest() {
    TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
    ArrayDeque<Integer> queue = new ArrayDeque<Integer>();
    AtomicLong state = new AtomicLong();
    BooleanSupplier isCancelled = new BooleanSupplier() {
        @Override
        public boolean getAsBoolean() throws Exception {
            return false;
        }
    };

    ts.onSubscribe(new BooleanSubscription());
    queue.offer(1);
    state.getAndIncrement();

    QueueDrainHelper.postComplete(ts, queue, state, isCancelled);

    ts.assertResult(1);
}
 

@Test
public void postCompleteCancelled() {
    final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
    ArrayDeque<Integer> queue = new ArrayDeque<Integer>();
    AtomicLong state = new AtomicLong();
    BooleanSupplier isCancelled = new BooleanSupplier() {
        @Override
        public boolean getAsBoolean() throws Exception {
            return ts.isCancelled();
        }
    };

    ts.onSubscribe(new BooleanSubscription());
    queue.offer(1);
    state.getAndIncrement();
    ts.cancel();

    QueueDrainHelper.postComplete(ts, queue, state, isCancelled);

    ts.assertEmpty();
}
 

@Test
public void postCompleteCancelledAfterOne() {
    final TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
        @Override
        public void onNext(Integer t) {
            super.onNext(t);
            cancel();
        }
    };
    ArrayDeque<Integer> queue = new ArrayDeque<Integer>();
    AtomicLong state = new AtomicLong();
    BooleanSupplier isCancelled = new BooleanSupplier() {
        @Override
        public boolean getAsBoolean() throws Exception {
            return ts.isCancelled();
        }
    };

    ts.onSubscribe(new BooleanSubscription());
    queue.offer(1);
    state.getAndIncrement();

    QueueDrainHelper.postComplete(ts, queue, state, isCancelled);

    ts.assertValue(1).assertNoErrors().assertNotComplete();
}
 

@Test
public void testMessagesToInactive() throws XMPPException, TigaseStringprepException {
    String recipient = "[email protected]";
    JID recp1 = JID.jidInstanceNS(recipient + "/res1");
    JID connId1 = JID.jidInstanceNS("[email protected]/recipient1-res1");
    XMPPResourceConnection session1 = getSession(connId1, recp1);

    enableCSI(session1);

    ArrayDeque<Packet> results = new ArrayDeque<>();
    Packet p = Packet.packetInstance("message", "[email protected]/res1", recp1.toString(), StanzaType.chat);
    p.setPacketTo(connId1);
    results.offer(p);
    Packet[] expected = results.toArray(new Packet[results.size()]);
    csi.filter(p, session1, null, results);
    Packet[] processed = results.toArray(new Packet[results.size()]);
    Assert.assertArrayEquals(expected, processed);
}
 

@Test
public void testPresenceToInactive() throws XMPPException, TigaseStringprepException {
    String recipient = "[email protected]";
    JID recp1 = JID.jidInstanceNS(recipient + "/res1");
    JID connId1 = JID.jidInstanceNS("[email protected]/recipient1-res1");
    XMPPResourceConnection session1 = getSession(connId1, recp1);

    enableCSI(session1);

    ArrayDeque<Packet> results = new ArrayDeque<>();
    Packet p = Packet.packetInstance("presence", "[email protected]/res1", recp1.toString(), StanzaType.available);
    p.setPacketTo(connId1);
    results.offer(p);
    Packet[] expected = new Packet[0];
    csi.filter(p, session1, null, results);
    Packet[] processed = results.toArray(new Packet[results.size()]);
    Assert.assertArrayEquals(expected, processed);
}
 

@Test
public void completeRequestRace() {
    for (int i = 0; i < 500; i++) {
        final TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
        final ArrayDeque<Integer> queue = new ArrayDeque<Integer>();
        final AtomicLong state = new AtomicLong();
        final BooleanSupplier isCancelled = new BooleanSupplier() {
            @Override
            public boolean getAsBoolean() throws Exception {
                return false;
            }
        };

        ts.onSubscribe(new BooleanSubscription());
        queue.offer(1);

        Runnable r1 = new Runnable() {
            @Override
            public void run() {
                QueueDrainHelper.postCompleteRequest(1, ts, queue, state, isCancelled);
            }
        };

        Runnable r2 = new Runnable() {
            @Override
            public void run() {
                QueueDrainHelper.postComplete(ts, queue, state, isCancelled);
            }
        };

        TestCommonHelper.race(r1, r2);

        ts.assertResult(1);
    }
}
 

/**
 * offer(null) throws NPE
 */
public void testOfferNull() {
    ArrayDeque q = new ArrayDeque();
    try {
        q.offer(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}
 

@Override
protected void deallocate(final ByteBuf buff) {
  final ArrayDeque<ByteBuf> queue = BUFFERS.get();
  if (!queue.offer(buff)) {
    buff.release();
  }
}
 

private ArrayList<Type<?>> sortTypes() {
    // sort the types in table creation order to avoid referencing not created table via a
    // reference (could also add constraints at the end but SQLite doesn't support that)
    ArrayDeque<Type<?>> queue = new ArrayDeque<>(model.getTypes());
    ArrayList<Type<?>> sorted = new ArrayList<>();
    while (!queue.isEmpty()) {
        Type<?> type = queue.poll();

        if (type.isView()) {
            continue;
        }

        Set<Type<?>> referencing = referencedTypesOf(type);
        for (Type<?> referenced : referencing) {
            Set<Type<?>> backReferences = referencedTypesOf(referenced);
            if (backReferences.contains(type)) {
                throw new CircularReferenceException("circular reference detected between "
                    + type.getName() + " and " + referenced.getName());
            }
        }
        if (referencing.isEmpty() || sorted.containsAll(referencing)) {
            sorted.add(type);
            queue.remove(type);
        } else {
            queue.offer(type); // put back
        }
    }
    return sorted;
}
 

@Override
public void onNext(T t) {
    ArrayDeque<T> bs = buffer;

    if (bs.size() == n) {
        bs.poll();
    }
    bs.offer(t);
}
 

/**
 * offer(null) throws NPE
 */
public void testOfferNull() {
    ArrayDeque q = new ArrayDeque();
    try {
        q.offer(null);
        shouldThrow();
    } catch (NullPointerException success) {}
}
 

@Test
public void testFlushStopping() throws XMPPException, TigaseStringprepException {
    String recipient = "[email protected]";
    JID recp1 = JID.jidInstanceNS(recipient + "/res1");
    JID connId1 = JID.jidInstanceNS("[email protected]/recipient1-res1");
    XMPPResourceConnection session1 = getSession(connId1, recp1);

    enableCSI(session1);

    ArrayDeque<Packet> results = new ArrayDeque<>();
    Packet p = Packet.packetInstance("presence", "[email protected]/res1", recp1.toString(), StanzaType.available);
    p.setPacketTo(connId1);
    csi.filter(p, session1, null, results);

    results.clear();
    Packet m = Packet.packetInstance("message", "[email protected]/res1", recp1.toString(), StanzaType.chat);
    m.getElement().addChild(new Element("received", new String[]{ "xmlns" }, new String[] { "urn:xmpp:receipts" }));
    m.setPacketTo(connId1);
    results.offer(m);
    csi.filter(m, session1, null, results);

    results.clear();
    results.offer(m);
    results.clear();
    csi.stopped(session1, results, new HashMap<>());
    Packet[] processed = results.toArray(new Packet[results.size()]);
    Assert.assertEquals(0, processed.length);
}
 

@Override
protected final void deallocate(final ByteBuffer buff) {
  buff.order(BIG_ENDIAN);
  final ArrayDeque<ByteBuffer> queue = BUFFERS.get();
  queue.offer(buff);
}
 

@Override
public boolean tryAdvance(Consumer<? super A> action) {
    Spliterator<T> source = this.source;
    if (source == null || localCancelled) {
        this.source = null;
        return false;
    }
    A acc = supplier.get();
    try {
        source.forEachRemaining(t -> {
            accumulator.accept(acc, t);
            if (cancelPredicate.test(acc)) {
                cancelSuffix();
                throw new CancelException();
            }
            if (localCancelled) {
                throw new CancelException();
            }
        });
    } catch (CancelException ex) {
        if (localCancelled) {
            return false;
        }
    }
    this.source = null;
    A result = acc;
    while (true) {
        if (prefix == null && suffix == null) {
            action.accept(result);
            return true;
        }
        ArrayDeque<A> res = new ArrayDeque<>();
        res.offer(result);
        synchronized (lock) {
            if (localCancelled)
                return false;
            OrderedCancellableSpliterator<T, A> s = prefix;
            while (s != null) {
                if (s.payload == null)
                    break;
                res.offerFirst(s.payload);
                s = s.prefix;
            }
            prefix = s;
            if (s != null) {
                s.suffix = this;
            }
            s = suffix;
            while (s != null) {
                if (s.payload == null)
                    break;
                res.offerLast(s.payload);
                s = s.suffix;
            }
            suffix = s;
            if (s != null) {
                s.prefix = this;
            }
            if (res.size() == 1) {
                if (prefix == null && suffix == null) {
                    action.accept(result);
                    return true;
                }
                this.payload = result;
                break;
            }
        }
        result = res.pollFirst();
        while (!res.isEmpty()) {
            result = combiner.apply(result, res.pollFirst());
            if (cancelPredicate.test(result)) {
                cancelSuffix();
            }
        }
    }
    return false;
}
 

@Override
public void onNext(T t) {

    ArrayDeque<T> bs = buffer;

    if (bs.size() == n) {
        T v = bs.poll();

        actual.onNext(v);
    }
    bs.offer(t);

}