TimelyDataProducer implements Runnable { public void collect(Map<String, Object> data, boolean asynchronous) { if (data != null && !data.isEmpty()) { String id = ""; if (asynchronous) { id = String.valueOf((Long) data.get(CollectionConstants.TASK_ID)); if (!CommonUtil.isBlank(id)) { id = CollectionConstants.ASYNCHRONIZATION_PREFIX + id; } } else { id = (String) data.get(CollectionConstants.OBJECT_ID); if (!CommonUtil.isBlank(id)) { id = CollectionConstants.SYNCHRONIZATION_PREFIX + id; } } data.remove(CollectionConstants.OBJECT_ID); data.put(CollectionConstants.OBJECT_ID, id); if (!CommonUtil.isBlank(id)) { for (Entry<String, ?> field : data.entrySet()) { dataContainer.push(id, field.getKey(), field.getValue()); } } } } TimelyDataProducer(DataContainer dataContainer, CollectionDriver driver); @Override void run(); void collect(Map<String, Object> data, boolean asynchronous); CollectOperationManager getCollectOperationManager(); void setCollectOperationManager(
         CollectOperationManager collectOperationManager);  }
@Test public void testCollect() { Map<String, Object> paramData = new HashMap<>(); long taskId = 100L; String resourceId = "1000"; paramData.put(CollectionConstants.OPERATION_NAME, CollectionConstants.METHOD_CREATE_CLUSTER); paramData.put(CollectionConstants.TASK_ID, taskId); TimelyDataProducer timelyDataProducer = new TimelyDataProducer(dataContainer, collectionDriver); timelyDataProducer.collect(paramData, true); String id = CollectionConstants.ASYNCHRONIZATION_PREFIX + taskId; verify(dataContainer).push(id ,CollectionConstants.OBJECT_ID, id); verify(dataContainer).push(id ,CollectionConstants.OPERATION_NAME, CollectionConstants.METHOD_CREATE_CLUSTER); verify(dataContainer).push(id ,CollectionConstants.TASK_ID, taskId); paramData.remove(CollectionConstants.TASK_ID); paramData.put(CollectionConstants.OBJECT_ID, resourceId); timelyDataProducer.collect(paramData, false); id = CollectionConstants.SYNCHRONIZATION_PREFIX + resourceId; verify(dataContainer).push(id ,CollectionConstants.OBJECT_ID, id); verify(dataContainer).push(id ,CollectionConstants.OPERATION_NAME, CollectionConstants.METHOD_CREATE_CLUSTER); }