Skip to content

Commit

Permalink
fix(pipeline): close chained pipelined objects
Browse files Browse the repository at this point in the history
  • Loading branch information
stakach committed Jul 3, 2023
1 parent d893af6 commit d649951
Show file tree
Hide file tree
Showing 2 changed files with 14 additions and 3 deletions.
2 changes: 1 addition & 1 deletion shard.yml
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
name: tasker
version: 2.1.3
version: 2.1.4
crystal: ">= 0.36.1"

dependencies:
Expand Down
15 changes: 13 additions & 2 deletions src/tasker/pipeline.cr
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
class Tasker
module Processor(Input)
abstract def process(input : Input) : Bool
abstract def close : Nil
abstract def closed? : Bool
end

class Subscription(Input)
Expand All @@ -13,6 +15,14 @@ class Tasker
@work.call input
true
end

def close : Nil
end

# check if the pipline is running
def closed? : Bool
false
end
end

# a lossy pipeline for realtime processing so any outputs are
Expand Down Expand Up @@ -81,12 +91,13 @@ class Tasker
end

# shutdown processing
def close
def close : Nil
@in.close
@chained.each(&.close)
end

# check if the pipline is running
def closed?
def closed? : Bool
@in.closed?
end

Expand Down

0 comments on commit d649951

Please sign in to comment.