@@ -129,6 +129,21 @@ impl DedupSink {
129
129
}
130
130
}
131
131
132
+ fn log_skipping_message ( & self , state : & mut DedupSinkState ) -> Result < ( ) > {
133
+ if state. skipped_count != 0 {
134
+ let last_record = state. last_record . as_ref ( ) . unwrap ( ) . as_ref ( ) ;
135
+ match state. skipped_count . cmp ( & 1 ) {
136
+ Ordering :: Equal => self . log_record ( & last_record) ?,
137
+ Ordering :: Greater => self . log_record (
138
+ & last_record
139
+ . replace_payload ( format ! ( "(skipped {} duplicates)" , state. skipped_count) ) ,
140
+ ) ?,
141
+ Ordering :: Less => unreachable ! ( ) , // We have checked if `state.skipped_count != 0`
142
+ }
143
+ }
144
+ Ok ( ( ) )
145
+ }
146
+
132
147
fn log_record ( & self , record : & Record ) -> Result < ( ) > {
133
148
#[ allow( clippy:: manual_try_fold) ] // https://github.com/rust-lang/rust-clippy/issues/11554
134
149
self . sinks . iter ( ) . fold ( Ok ( ( ) ) , |result, sink| {
@@ -152,18 +167,7 @@ impl Sink for DedupSink {
152
167
state. skipped_count += 1 ;
153
168
return Ok ( ( ) ) ;
154
169
}
155
-
156
- if state. skipped_count != 0 {
157
- let last_record = state. last_record . as_ref ( ) . unwrap ( ) . as_ref ( ) ;
158
- match state. skipped_count . cmp ( & 1 ) {
159
- Ordering :: Equal => self . log_record ( & last_record) ?,
160
- Ordering :: Greater => self . log_record (
161
- & last_record
162
- . replace_payload ( format ! ( "(skipped {} duplicates)" , state. skipped_count) ) ,
163
- ) ?,
164
- Ordering :: Less => unreachable ! ( ) , // We have checked if `state.skipped_count != 0`
165
- }
166
- }
170
+ self . log_skipping_message ( & mut state) ?;
167
171
168
172
self . log_record ( record) ?;
169
173
state. skipped_count = 0 ;
@@ -179,6 +183,17 @@ impl Sink for DedupSink {
179
183
helper:: common_impl!( @Sink : common_impl) ;
180
184
}
181
185
186
+ impl Drop for DedupSink {
187
+ fn drop ( & mut self ) {
188
+ if let Err ( err) = self . log_skipping_message ( & mut self . state . lock_expect ( ) ) {
189
+ self . common_impl . non_returnable_error ( "DedupSink" , err) ;
190
+ }
191
+ if let Err ( err) = self . flush_sinks ( ) {
192
+ self . common_impl . non_returnable_error ( "DedupSink" , err) ;
193
+ }
194
+ }
195
+ }
196
+
182
197
/// #
183
198
#[ doc = include_str ! ( "../include/doc/generic-builder-note.md" ) ]
184
199
pub struct DedupSinkBuilder < ArgS > {
@@ -330,4 +345,64 @@ mod tests {
330
345
assert_eq ! ( records[ 12 ] . payload( ) , "Meow~ Meow..." ) ;
331
346
assert_eq ! ( records[ 12 ] . level( ) , Level :: Info ) ;
332
347
}
348
+
349
+ #[ test]
350
+ fn dedup_on_drop ( ) {
351
+ {
352
+ let records = {
353
+ let test_sink = Arc :: new ( TestSink :: new ( ) ) ;
354
+ {
355
+ let dedup_sink = Arc :: new (
356
+ DedupSink :: builder ( )
357
+ . skip_duration ( Duration :: from_secs ( 1 ) )
358
+ . sink ( test_sink. clone ( ) )
359
+ . build ( )
360
+ . unwrap ( ) ,
361
+ ) ;
362
+ let test = build_test_logger ( |b| b. sink ( dedup_sink) ) ;
363
+
364
+ info ! ( logger: test, "I wish I was a cat" ) ;
365
+ info ! ( logger: test, "I wish I was a cat" ) ;
366
+ }
367
+ test_sink. records ( )
368
+ } ;
369
+
370
+ assert_eq ! ( records. len( ) , 2 ) ;
371
+
372
+ assert_eq ! ( records[ 0 ] . payload( ) , "I wish I was a cat" ) ;
373
+ assert_eq ! ( records[ 0 ] . level( ) , Level :: Info ) ;
374
+
375
+ assert_eq ! ( records[ 1 ] . payload( ) , "I wish I was a cat" ) ;
376
+ assert_eq ! ( records[ 1 ] . level( ) , Level :: Info ) ;
377
+ }
378
+
379
+ {
380
+ let records = {
381
+ let test_sink = Arc :: new ( TestSink :: new ( ) ) ;
382
+ {
383
+ let dedup_sink = Arc :: new (
384
+ DedupSink :: builder ( )
385
+ . skip_duration ( Duration :: from_secs ( 1 ) )
386
+ . sink ( test_sink. clone ( ) )
387
+ . build ( )
388
+ . unwrap ( ) ,
389
+ ) ;
390
+ let test = build_test_logger ( |b| b. sink ( dedup_sink) ) ;
391
+
392
+ info ! ( logger: test, "I wish I was a cat" ) ;
393
+ info ! ( logger: test, "I wish I was a cat" ) ;
394
+ info ! ( logger: test, "I wish I was a cat" ) ;
395
+ }
396
+ test_sink. records ( )
397
+ } ;
398
+
399
+ assert_eq ! ( records. len( ) , 2 ) ;
400
+
401
+ assert_eq ! ( records[ 0 ] . payload( ) , "I wish I was a cat" ) ;
402
+ assert_eq ! ( records[ 0 ] . level( ) , Level :: Info ) ;
403
+
404
+ assert_eq ! ( records[ 1 ] . payload( ) , "(skipped 2 duplicates)" ) ;
405
+ assert_eq ! ( records[ 1 ] . level( ) , Level :: Info ) ;
406
+ }
407
+ }
333
408
}
0 commit comments