2525import java .io .BufferedReader ;
2626import java .io .BufferedWriter ;
2727import java .io .IOException ;
28+ import java .io .UncheckedIOException ;
2829import java .nio .file .Files ;
2930import java .nio .file .Path ;
31+ import java .util .HashSet ;
3032import java .util .Map ;
3133import java .util .Objects ;
34+ import java .util .Set ;
3235import java .util .concurrent .ConcurrentHashMap ;
3336import java .util .concurrent .ConcurrentMap ;
3437import java .util .concurrent .TimeUnit ;
3538import java .util .concurrent .atomic .LongAdder ;
39+ import java .util .function .BiFunction ;
40+ import java .util .function .Function ;
41+ import java .util .function .Supplier ;
3642import java .util .stream .Collectors ;
43+ import java .util .stream .Stream ;
44+
45+ import static org .radarcns .hdfs .accounting .OffsetRangeFile .COMMA_PATTERN ;
46+ import static org .radarcns .hdfs .accounting .OffsetRangeFile .read ;
47+ import static org .radarcns .hdfs .util .ThrowingConsumer .tryCatch ;
3748
3849/** Store overview of records written, divided into bins. */
3950public class BinFile extends PostponedWriter {
4051 private static final Logger logger = LoggerFactory .getLogger (BinFile .class );
52+ private static final String BINS_HEADER = String .join (
53+ "," , "topic" , "device" , "timestamp" , "count" ) + "\n " ;
4154
42- private final ConcurrentMap <Bin , LongAdder > bins ;
55+ private final ConcurrentMap <Bin , Long > bins ;
4356 private final Path path ;
4457 private final StorageDriver storage ;
4558
46- public BinFile (@ Nonnull StorageDriver storageDriver , @ Nonnull Path path ,
47- @ Nonnull ConcurrentMap <Bin , LongAdder > initialData ) {
59+ public BinFile (@ Nonnull StorageDriver storageDriver , @ Nonnull Path path ) {
4860 super ("bins" , 5 , TimeUnit .SECONDS );
4961 Objects .requireNonNull (path );
50- Objects .requireNonNull (initialData );
5162 this .storage = storageDriver ;
5263 this .path = path ;
53- this .bins = initialData ;
54- }
55-
56- public static BinFile read (StorageDriver storage , Path path ) {
57- ConcurrentMap <Bin , LongAdder > map = new ConcurrentHashMap <>();
58- try (BufferedReader input = storage .newBufferedReader (path )){
59- // Read in all lines as multikeymap (key, key, key, value)
60- String line = input .readLine ();
61- if (line != null ) {
62- line = input .readLine ();
63- while (line != null ) {
64- String [] columns = line .split ("," );
65- try {
66- Bin bin = new Bin (columns [0 ], columns [1 ], columns [2 ]);
67- LongAdder adder = new LongAdder ();
68- adder .add (Long .valueOf (columns [3 ]));
69- map .put (bin , adder );
70- } catch (ArrayIndexOutOfBoundsException ex ) {
71- logger .warn ("Unable to read row of the bins file. Skipping." );
72- }
73- line = input .readLine ();
74- }
75- }
76- } catch (IOException e ) {
77- logger .warn ("Could not read the file with bins. Creating new file when writing." );
78- }
79- return new BinFile (storage , path , map );
64+ this .bins = new ConcurrentHashMap <>();
8065 }
8166
8267 /** Add number of instances to given bin. */
8368 public void add (Bin bin , long value ) {
84- bins .computeIfAbsent (bin , b -> new LongAdder ()). add ( value );
69+ bins .compute (bin , compute (() -> 0L , v -> v + value ) );
8570 }
8671
8772 /** Put a map of bins. */
@@ -92,38 +77,87 @@ public void putAll(Map<? extends Bin, ? extends Number> binMap) {
9277 @ Override
9378 public String toString () {
9479 return bins .entrySet ().stream ()
95- .map (e -> e .getKey () + " - " + e .getValue (). sum () )
80+ .map (e -> e .getKey () + " - " + e .getValue ())
9681 .collect (Collectors .joining ("\n " ));
9782 }
9883
9984 @ Override
10085 protected void doWrite () {
86+ Path tempPath ;
10187 try {
102- Path tempPath = createTempFile ("bins" , ".csv" );
103-
104- // Write all bins to csv
105- try (BufferedWriter bw = Files .newBufferedWriter (tempPath )) {
106- String header = String .join ("," , "topic" , "device" , "timestamp" , "count" );
107- bw .write (header );
108- bw .write ('\n' );
109-
110- for (Map .Entry <Bin , LongAdder > entry : bins .entrySet ()) {
111- Bin bin = entry .getKey ();
112- bw .write (bin .getTopic ());
113- bw .write (',' );
114- bw .write (bin .getCategory ());
115- bw .write (',' );
116- bw .write (bin .getTime ());
117- bw .write (',' );
118- bw .write (String .valueOf (entry .getValue ().sum ()));
119- bw .write ('\n' );
120- }
88+ tempPath = createTempFile ("bins" , ".csv" );
89+ } catch (IOException e ) {
90+ logger .error ("Cannot create temporary bins file: {}" , e .toString ());
91+ return ;
92+ }
93+
94+ BufferedReader reader = null ;
95+ Stream <String []> lines ;
96+
97+ try {
98+ reader = storage .newBufferedReader (path );
99+
100+ if (reader .readLine () == null ) {
101+ lines = Stream .empty ();
102+ } else {
103+ lines = reader .lines ()
104+ .map (COMMA_PATTERN ::split );
121105 }
106+ } catch (IOException ex ){
107+ logger .warn ("Could not read the file with bins. Creating new file when writing." );
108+ lines = Stream .empty ();
109+ }
110+
111+ Set <Bin > binKeys = new HashSet <>(bins .keySet ());
112+
113+ try (BufferedWriter bw = Files .newBufferedWriter (tempPath )) {
114+ bw .write (BINS_HEADER );
115+
116+ lines .forEach (s -> {
117+ Bin bin = new Bin (s [0 ], s [1 ], s [2 ]);
118+ long value = Long .parseLong (s [3 ]);
119+ if (binKeys .remove (bin )) {
120+ value += bins .remove (bin );
121+ }
122+ writeLine (bw , bin , value );
123+ });
124+ binKeys .forEach (bin -> writeLine (bw , bin , bins .remove (bin )));
122125
123126 storage .store (tempPath , path );
124- } catch (IOException e ) {
127+ } catch (UncheckedIOException | IOException e ) {
125128 logger .error ("Failed to write bins: {}" , e );
129+ } finally {
130+ if (reader != null ) {
131+ try {
132+ reader .close ();
133+ } catch (IOException e ) {
134+ logger .debug ("Failed to close bin file reader" , e );
135+ }
136+ }
126137 }
127138 }
128139
140+ private static void writeLine (BufferedWriter writer , Bin bin , long value ) {
141+ try {
142+ writer .write (bin .getTopic ());
143+ writer .write (',' );
144+ writer .write (bin .getCategory ());
145+ writer .write (',' );
146+ writer .write (bin .getTime ());
147+ writer .write (',' );
148+ writer .write (Long .toString (value ));
149+ writer .write ('\n' );
150+ } catch (IOException e ) {
151+ throw new UncheckedIOException (e );
152+ }
153+ }
154+
155+ private static <K , V1 , V2 > BiFunction <K , V1 , V2 > compute (Supplier <? extends V1 > init , Function <? super V1 , ? extends V2 > update ) {
156+ return (k , v ) -> {
157+ if (v == null ) {
158+ v = init .get ();
159+ }
160+ return update .apply (v );
161+ };
162+ }
129163}
0 commit comments