@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); }
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 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 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); }
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); }
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);  }