@@ -269,7 +269,7 @@ pub mod source {
269
269
use std:: sync:: Arc ;
270
270
use timely:: dataflow:: { Scope , Stream , operators:: { Capability , CapabilitySet } } ;
271
271
use timely:: progress:: Timestamp ;
272
- use timely:: scheduling:: { SyncActivator , activate :: SyncActivateOnDrop } ;
272
+ use timely:: scheduling:: SyncActivator ;
273
273
274
274
// TODO(guswynn): implement this generally in timely
275
275
struct DropActivator {
@@ -351,14 +351,14 @@ pub mod source {
351
351
let address = messages_op. operator_info ( ) . address ;
352
352
let activator = scope. sync_activator_for ( & address) ;
353
353
let activator2 = scope. activator_for ( & address) ;
354
- let drop_activator = Arc :: new ( SyncActivateOnDrop :: new ( ( ) , scope. sync_activator_for ( & address) ) ) ;
354
+ let drop_activator = DropActivator { activator : Arc :: new ( scope. sync_activator_for ( & address) ) } ;
355
355
let mut source = source_builder ( activator) ;
356
356
let ( mut updates_out, updates) = messages_op. new_output ( ) ;
357
357
let ( mut progress_out, progress) = messages_op. new_output ( ) ;
358
358
messages_op. build ( |capabilities| {
359
359
360
360
// A Weak that communicates whether the returned token has been dropped.
361
- let drop_activator_weak = Arc :: downgrade ( & drop_activator) ;
361
+ let drop_activator_weak = Arc :: downgrade ( & drop_activator. activator ) ;
362
362
363
363
token = Some ( drop_activator) ;
364
364
0 commit comments