@SuppressWarnings("unchecked") @Test public void nonSyncProcessing() { DefaultData defaultData = mock(DefaultData.class); AbstractDataProcessor dataProcessor = mock(AbstractDataProcessor.class); Future<Void> future1 = mock(Future.class); Future<Void> future2 = mock(Future.class); when(dataProcessor.process(defaultData)).thenReturn(Collections.singletonList(future1)); when(dataProcessor.flush()).thenReturn(Collections.singletonList(future2)); Collection<Future<Void>> futures = storageWriter.process(Collections.singletonList(defaultData), Collections.singletonList(dataProcessor)); verify(dataProcessor, times(1)).setStorageWriter(storageWriter); verify(dataProcessor, times(1)).process(defaultData); verify(dataProcessor, times(1)).flush(); verify(dataProcessor, times(1)).setStorageWriter(null); assertThat(futures, hasSize(2)); assertThat(futures, hasItem(future1)); assertThat(futures, hasItem(future2)); }
public Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors) { List<Future<Void>> futureList = new ArrayList<>(); if ((null != processors) && !processors.isEmpty()) { for (AbstractDataProcessor processor : processors) { processor.setStorageWriter(this); } for (DefaultData defaultData : defaultDataList) { for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.process(defaultData)); } } for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.flush()); processor.setStorageWriter(null); } } else { for (DefaultData defaultData : defaultDataList) { Future<Void> future = this.write(defaultData); futureList.add(future); } } return futureList; }
StorageWriter implements IWriter { public Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors) { List<Future<Void>> futureList = new ArrayList<>(); if ((null != processors) && !processors.isEmpty()) { for (AbstractDataProcessor processor : processors) { processor.setStorageWriter(this); } for (DefaultData defaultData : defaultDataList) { for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.process(defaultData)); } } for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.flush()); processor.setStorageWriter(null); } } else { for (DefaultData defaultData : defaultDataList) { Future<Void> future = this.write(defaultData); futureList.add(future); } } return futureList; } }
StorageWriter implements IWriter { public Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors) { List<Future<Void>> futureList = new ArrayList<>(); if ((null != processors) && !processors.isEmpty()) { for (AbstractDataProcessor processor : processors) { processor.setStorageWriter(this); } for (DefaultData defaultData : defaultDataList) { for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.process(defaultData)); } } for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.flush()); processor.setStorageWriter(null); } } else { for (DefaultData defaultData : defaultDataList) { Future<Void> future = this.write(defaultData); futureList.add(future); } } return futureList; }  }
StorageWriter implements IWriter { public Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors) { List<Future<Void>> futureList = new ArrayList<>(); if ((null != processors) && !processors.isEmpty()) { for (AbstractDataProcessor processor : processors) { processor.setStorageWriter(this); } for (DefaultData defaultData : defaultDataList) { for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.process(defaultData)); } } for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.flush()); processor.setStorageWriter(null); } } else { for (DefaultData defaultData : defaultDataList) { Future<Void> future = this.write(defaultData); futureList.add(future); } } return futureList; }  Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors); void processSynchronously(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors); @Override Future<Void> write(DefaultData defaultData); @Override Future<Void> write(DefaultData defaultData, Map<?, ?> kryoPreferences); synchronized boolean prepareForWrite(StorageData storageData); final synchronized void cancel(); final synchronized void closeStorageWriter(); long getQueuedTaskCount(); boolean writeNonDefaultDataObject(Object object, String fileName); Path getWritingFolderPath(); String getExecutorServiceStatus(); boolean isWritingOn(); StorageData getStorageData(); WritingStatus getWritingStatus(); @PostConstruct void postConstruct(); @Override String toString(); }
StorageWriter implements IWriter { public Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors) { List<Future<Void>> futureList = new ArrayList<>(); if ((null != processors) && !processors.isEmpty()) { for (AbstractDataProcessor processor : processors) { processor.setStorageWriter(this); } for (DefaultData defaultData : defaultDataList) { for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.process(defaultData)); } } for (AbstractDataProcessor processor : processors) { futureList.addAll(processor.flush()); processor.setStorageWriter(null); } } else { for (DefaultData defaultData : defaultDataList) { Future<Void> future = this.write(defaultData); futureList.add(future); } } return futureList; }  Collection<Future<Void>> process(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors); void processSynchronously(Collection<? extends DefaultData> defaultDataList, Collection<AbstractDataProcessor> processors); @Override Future<Void> write(DefaultData defaultData); @Override Future<Void> write(DefaultData defaultData, Map<?, ?> kryoPreferences); synchronized boolean prepareForWrite(StorageData storageData); final synchronized void cancel(); final synchronized void closeStorageWriter(); long getQueuedTaskCount(); boolean writeNonDefaultDataObject(Object object, String fileName); Path getWritingFolderPath(); String getExecutorServiceStatus(); boolean isWritingOn(); StorageData getStorageData(); WritingStatus getWritingStatus(); @PostConstruct void postConstruct(); @Override String toString();  }