github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 22:57:16 -04:00
Code Issues Projects Releases Packages Wiki Activity

use method-level ensure blocks

Browse source 
Fixes #10872
This commit is contained in:
Dan Hermann 2019-06-28 07:12:15 -05:00
parent 1802869d80
commit 46c1611dff
1 changed files with 44 additions and 50 deletions
Download patch file Download diff file Expand all files Collapse all files

94
logstash-core/lib/logstash/pipelines_registry.rb
View file

@ -48,29 +48,27 @@ module LogStash
# @return [Boolean] new pipeline creation success
def create_pipeline(pipeline_id, pipeline, &create_block)
lock = get_lock(pipeline_id)
begin
lock.lock
lock.lock
success = false
success = false
state = @states.get(pipeline_id)
if state
if state.terminated?
success = yield
state.set_pipeline(pipeline)
else
logger.error("Attempted to create a pipeline that already exists", :pipeline_id => pipeline_id)
end
@states.put(pipeline_id, state)
else
state = @states.get(pipeline_id)
if state
if state.terminated?
success = yield
@states.put(pipeline_id, PipelineState.new(pipeline_id, pipeline)) if success
state.set_pipeline(pipeline)
else
logger.error("Attempted to create a pipeline that already exists", :pipeline_id => pipeline_id)
end
success
ensure
lock.unlock
@states.put(pipeline_id, state)
else
success = yield
@states.put(pipeline_id, PipelineState.new(pipeline_id, pipeline)) if success
end
success
ensure
lock.unlock
end
# Execute the passed termination logic block
@ -80,20 +78,18 @@ module LogStash
# @yieldparam [Pipeline] the pipeline to terminate
def terminate_pipeline(pipeline_id, &stop_block)
lock = get_lock(pipeline_id)
begin
lock.lock
lock.lock
state = @states.get(pipeline_id)
if state.nil?
logger.error("Attempted to terminate a pipeline that does not exists", :pipeline_id => pipeline_id)
@states.remove(pipeline_id)
else
yield(state.pipeline)
@states.put(pipeline_id, state)
end
ensure
lock.unlock
state = @states.get(pipeline_id)
if state.nil?
logger.error("Attempted to terminate a pipeline that does not exists", :pipeline_id => pipeline_id)
@states.remove(pipeline_id)
else
yield(state.pipeline)
@states.put(pipeline_id, state)
end
ensure
lock.unlock
end
# Execute the passed reloading logic block in the context of the reloading state and set new pipeline in state
@ -105,29 +101,27 @@ module LogStash
# @return [Boolean] new pipeline creation success
def reload_pipeline(pipeline_id, &reload_block)
lock = get_lock(pipeline_id)
begin
lock.lock
success = false
lock.lock
success = false
state = @states.get(pipeline_id)
if state.nil?
logger.error("Attempted to reload a pipeline that does not exists", :pipeline_id => pipeline_id)
@states.remove(pipeline_id)
else
state.set_reloading(true)
begin
success, new_pipeline = yield
state.set_pipeline(new_pipeline)
ensure
state.set_reloading(false)
end
@states.put(pipeline_id, state)
state = @states.get(pipeline_id)
if state.nil?
logger.error("Attempted to reload a pipeline that does not exists", :pipeline_id => pipeline_id)
@states.remove(pipeline_id)
else
state.set_reloading(true)
begin
success, new_pipeline = yield
state.set_pipeline(new_pipeline)
ensure
state.set_reloading(false)
end
success
ensure
lock.unlock
@states.put(pipeline_id, state)
end
success
ensure
lock.unlock
end
# @param pipeline_id [String, Symbol] the pipeline id