Class: Operaton::Bpm::Client::Topic::Impl::TopicSubscriptionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb

Overview

Mirrors org.operaton.bpm.client.topic.impl.TopicSubscriptionManager. Runs the fetch-and-lock loop on a dedicated thread.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(engine_client, typed_values, client_lock_duration) ⇒ TopicSubscriptionManager

Returns a new instance of TopicSubscriptionManager.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 23

def initialize(engine_client, typed_values, client_lock_duration)
  @engine_client = engine_client
  @subscriptions = []
  @subscriptions_mutex = Mutex.new
  @task_topic_requests = []
  @external_task_handlers = {}
  @client_lock_duration = client_lock_duration
  @typed_values = typed_values
  @external_task_service = Task::Impl::ExternalTaskServiceImpl.new(engine_client)
  @is_backoff_strategy_disabled = false
  @backoff_strategy = nil

  @is_running = false
  @state_mutex = Mutex.new
  @acquisition_monitor = Mutex.new
  @is_waiting = ConditionVariable.new
  @resume_signaled = false
  @thread = nil
end

Instance Attribute Details

#engine_client ⇒ Object (readonly)

Returns the value of attribute engine_client.



21
22
23
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 21

def engine_client
  @engine_client
end

Instance Method Details

#backoff_strategy=(backoff_strategy) ⇒ Object 



101
102
103
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 101

def backoff_strategy=(backoff_strategy)
  @backoff_strategy = backoff_strategy
end

#disable_backoff_strategy ⇒ Object 



105
106
107
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 105

def disable_backoff_strategy
  @is_backoff_strategy_disabled = true
end

#run ⇒ Object 



43
44
45
46
47
48
49
50
51
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 43

def run
  while running?
    begin
      acquire
    rescue StandardError => e
      logger.exception_while_acquiring_tasks(e)
    end
  end
end

#running? ⇒ Boolean Also known as: is_running?

Returns:

  • (Boolean) 


96
97
98
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 96

def running?
  @state_mutex.synchronize { @is_running }
end

#start ⇒ Object 



67
68
69
70
71
72
73
74
75
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 67

def start
  @state_mutex.synchronize do
    return if @is_running

    @is_running = true
    @thread = Thread.new { run }
    @thread.name = "TopicSubscriptionManager"
  end
end

#stop ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 53

def stop
  @state_mutex.synchronize do
    return unless @is_running

    @is_running = false
  end
  resume
  begin
    @thread&.join
  rescue StandardError => e
    logger.exception_while_shutting_down(e)
  end
end

#subscribe(subscription) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 77

def subscribe(subscription)
  @subscriptions_mutex.synchronize do
    if @subscriptions.include?(subscription)
      raise logger.topic_name_already_subscribed_exception(subscription.topic_name)
    end

    @subscriptions << subscription
  end
  resume
end

#subscriptions ⇒ Object 



92
93
94
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 92

def subscriptions
  @subscriptions_mutex.synchronize { @subscriptions.dup }
end

#unsubscribe(subscription) ⇒ Object 



88
89
90
 
# File 'lib/operaton/bpm/client/topic/impl/topic_subscription_manager.rb', line 88

def unsubscribe(subscription)
  @subscriptions_mutex.synchronize { @subscriptions.delete(subscription) }
end