test_class.declarator: topicName = "com.weevent.test"
test_class.modifier: private
test_class.original_string: private String topicName = "com.weevent.test";
test_class.type: String
test_class.var_name: topicName
test_class.declarator: subscriptionId = "abc"
test_class.modifier: private
test_class.original_string: private String subscriptionId = "abc";
test_class.type: String
test_class.var_name: subscriptionId
test_class.declarator: threadPoolTaskExecutor
test_class.modifier: private
test_class.original_string: private ThreadPoolTaskExecutor threadPoolTaskExecutor;
test_class.type: ThreadPoolTaskExecutor
test_class.var_name: threadPoolTaskExecutor
test_class.declarator: wait3s = 3000
test_class.modifier: private final
test_class.original_string: private final long wait3s = 3000;
test_class.type: long
test_class.var_name: wait3s
test_class.file: weevent-core/src/test/java/com/webank/weevent/core/task/NotifyTaskTest.java
test_class.identifier: NotifyTaskTest
test_class.superclass: extends JUnitTestBase
test_case.body: @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_case.class_method_signature: NotifyTaskTest.testPush()
test_case.full_signature: @Test public void testPush()
test_case.identifier: testPush
test_case.invocations: execute add newEvent push sleep assertEquals assertEquals assertEquals getNotifiedCount
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testPush()
focal_class.declarator: subscriptionId
focal_class.modifier: private final
focal_class.original_string: private final String subscriptionId;
focal_class.type: String
focal_class.var_name: subscriptionId
focal_class.declarator: consumerListener
focal_class.modifier: private final
focal_class.original_string: private final IConsumer.ConsumerListener consumerListener;
focal_class.type: IConsumer.ConsumerListener
focal_class.var_name: consumerListener
focal_class.declarator: idleTime
focal_class.modifier: private final
focal_class.original_string: private final int idleTime;
focal_class.type: int
focal_class.var_name: idleTime
focal_class.declarator: eventQueue = new LinkedBlockingDeque<>()
focal_class.modifier: private final
focal_class.original_string: private final BlockingDeque<WeEvent> eventQueue = new LinkedBlockingDeque<>();
focal_class.type: BlockingDeque<WeEvent>
focal_class.var_name: eventQueue
focal_class.declarator: notifiedCount = 0
focal_class.modifier: private
focal_class.original_string: private long notifiedCount = 0;
focal_class.type: long
focal_class.var_name: notifiedCount
focal_class.declarator: lastTimeStamp = new Date()
focal_class.modifier: private final
focal_class.original_string: private final Date lastTimeStamp = new Date();
focal_class.type: Date
focal_class.var_name: lastTimeStamp
focal_class.declarator: mergeCache = new FixedFIFOCache<>(1024)
focal_class.modifier: private final
focal_class.original_string: private final Map<String, Long> mergeCache = new FixedFIFOCache<>(1024);
focal_class.type: Map<String, Long>
focal_class.var_name: mergeCache
focal_class.file: weevent-core/src/main/java/com/webank/weevent/core/task/NotifyTask.java
focal_class.identifier: NotifyTask
focal_class.class_method_signature: NotifyTask.NotifyTask(String subscriptionId, int idleTime, @NonNull IConsumer.ConsumerListener consumerListener)
focal_class.full_signature: public  NotifyTask(String subscriptionId, int idleTime, @NonNull IConsumer.ConsumerListener consumerListener)
focal_class.identifier: NotifyTask
focal_class.modifiers: public
focal_class.parameters: (String subscriptionId, int idleTime, @NonNull IConsumer.ConsumerListener consumerListener)
focal_class.signature:  NotifyTask(String subscriptionId, int idleTime, @NonNull IConsumer.ConsumerListener consumerListener)
focal_class.class_method_signature: NotifyTask.push(List<WeEvent> events)
focal_class.full_signature: public void push(List<WeEvent> events)
focal_class.identifier: push
focal_class.modifiers: public
focal_class.parameters: (List<WeEvent> events)
focal_class.return: void
focal_class.signature: void push(List<WeEvent> events)
focal_class.class_method_signature: NotifyTask.taskOnceLoop()
focal_class.full_signature: @Override protected void taskOnceLoop()
focal_class.identifier: taskOnceLoop
focal_class.modifiers: @Override protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void taskOnceLoop()
focal_class.superclass: extends StoppableTask
focal_method.body: public void push(List<WeEvent> events) {
          try {
              // offer event into queue one by one
              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();
          }
      }
focal_method.class_method_signature: NotifyTask.push(List<WeEvent> events)
focal_method.full_signature: public void push(List<WeEvent> events)
focal_method.identifier: push
focal_method.invocations: offer error onException debug error interrupt currentThread
focal_method.modifiers: public
focal_method.parameters: (List<WeEvent> events)
focal_method.return: void
focal_method.signature: void push(List<WeEvent> events)