### Question:
NotifyTask extends StoppableTask { public void push(List<WeEvent> events) { try { for (WeEvent event : events) { if (!this.eventQueue.offer(event, this.idleTime, TimeUnit.MILLISECONDS)) { log.error("push notify failed due to queue is full"); this.consumerListener.onException(new BrokerException(ErrorCode.SUBSCRIPTION_NOTIFY_QUEUE_FULL)); return; } log.debug("offer notify queue, event: {}", event); } } catch (InterruptedException e) { log.error("offer notify queue failed", e); Thread.currentThread().interrupt(); } } NotifyTask(String subscriptionId, int idleTime, @NonNull IConsumer.ConsumerListener consumerListener); void push(List<WeEvent> events);  }### Answer:
@Test public void testPush() throws Exception { MyListener listener = new MyListener(); NotifyTask notifyTask = new NotifyTask(this.subscriptionId, 1000, listener); this.threadPoolTaskExecutor.execute(notifyTask); List<WeEvent> data = new ArrayList<>(); data.add(this.newEvent("a")); notifyTask.push(data); Thread.sleep(wait3s); Assert.assertEquals(listener.received, 1); Assert.assertEquals(subscriptionId, listener.subscriptionId); Assert.assertEquals(notifyTask.getNotifiedCount(), listener.received); }
@Test public void testPush2() throws Exception { MyListener listener = new MyListener(); NotifyTask notifyTask = new NotifyTask(this.subscriptionId, 1000, listener); this.threadPoolTaskExecutor.execute(notifyTask); List<WeEvent> data = new ArrayList<>(); data.add(this.newEvent("a")); data.add(this.newEvent("b")); notifyTask.push(data); Thread.sleep(wait3s); Assert.assertEquals(listener.received, 2); Assert.assertEquals(subscriptionId, listener.subscriptionId); Assert.assertEquals(notifyTask.getNotifiedCount(), listener.received); }
@Test public void testPush3() throws Exception { MyListener listener = new MyListener(); NotifyTask notifyTask = new NotifyTask(this.subscriptionId, 1000, listener); this.threadPoolTaskExecutor.execute(notifyTask); List<WeEvent> data1 = new ArrayList<>(); data1.add(this.newEvent("a")); data1.add(this.newEvent("b")); notifyTask.push(data1); List<WeEvent> data2 = new ArrayList<>(); data2.add(this.newEvent("c")); notifyTask.push(data2); Thread.sleep(wait3s); Assert.assertEquals(listener.received, 3); Assert.assertEquals(subscriptionId, listener.subscriptionId); Assert.assertEquals(notifyTask.getNotifiedCount(), listener.received); }