channel.rb 9.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. # Copyright (C) 2012-2016 Zammad Foundation, http://zammad-foundation.org/
  2. class Channel < ApplicationModel
  3. include Channel::Assets
  4. belongs_to :group
  5. store :options
  6. store :preferences
  7. after_create :email_address_check
  8. after_update :email_address_check
  9. after_destroy :email_address_check
  10. # rubocop:disable Style/ClassVars
  11. @@channel_stream = {}
  12. @@channel_stream_started_till_at = {}
  13. # rubocop:enable Style/ClassVars
  14. =begin
  15. fetch all accounts
  16. Channel.fetch
  17. =end
  18. def self.fetch
  19. channels = Channel.where('active = ? AND area LIKE ?', true, '%::Account')
  20. channels.each(&:fetch)
  21. end
  22. =begin
  23. fetch one account
  24. channel = Channel.where(area: 'Email::Account').first
  25. channel.fetch
  26. =end
  27. def fetch(force = false)
  28. adapter = options[:adapter]
  29. adapter_options = options
  30. if options[:inbound] && options[:inbound][:adapter]
  31. adapter = options[:inbound][:adapter]
  32. adapter_options = options[:inbound][:options]
  33. end
  34. begin
  35. # we need to require each channel backend individually otherwise we get a
  36. # 'warning: toplevel constant Twitter referenced by Channel::Driver::Twitter' error e.g.
  37. # so we have to convert the channel name to the filename via Rails String.underscore
  38. # http://stem.ps/rails/2015/01/25/ruby-gotcha-toplevel-constant-referenced-by.html
  39. require "channel/driver/#{adapter.to_filename}"
  40. driver_class = Object.const_get("Channel::Driver::#{adapter.to_classname}")
  41. driver_instance = driver_class.new
  42. return if !force && !driver_instance.fetchable?(self)
  43. result = driver_instance.fetch(adapter_options, self)
  44. self.status_in = result[:result]
  45. self.last_log_in = result[:notice]
  46. preferences[:last_fetch] = Time.zone.now
  47. save!
  48. rescue => e
  49. error = "Can't use Channel::Driver::#{adapter.to_classname}: #{e.inspect}"
  50. logger.error error
  51. logger.error e.backtrace
  52. self.status_in = 'error'
  53. self.last_log_in = error
  54. preferences[:last_fetch] = Time.zone.now
  55. save!
  56. end
  57. end
  58. =begin
  59. stream instance of account
  60. channel = Channel.where(area: 'Twitter::Account').first
  61. stream_instance = channel.stream_instance
  62. # start stream
  63. stream_instance.stream
  64. =end
  65. def stream_instance
  66. adapter = options[:adapter]
  67. begin
  68. # we need to require each channel backend individually otherwise we get a
  69. # 'warning: toplevel constant Twitter referenced by Channel::Driver::Twitter' error e.g.
  70. # so we have to convert the channel name to the filename via Rails String.underscore
  71. # http://stem.ps/rails/2015/01/25/ruby-gotcha-toplevel-constant-referenced-by.html
  72. require "channel/driver/#{adapter.to_filename}"
  73. driver_class = Object.const_get("Channel::Driver::#{adapter.to_classname}")
  74. driver_instance = driver_class.new
  75. # check is stream exists
  76. return if !driver_instance.respond_to?(:stream_instance)
  77. driver_instance.stream_instance(self)
  78. # set scheduler job to active
  79. return driver_instance
  80. rescue => e
  81. error = "Can't use Channel::Driver::#{adapter.to_classname}: #{e.inspect}"
  82. logger.error error
  83. logger.error e.backtrace
  84. self.status_in = 'error'
  85. self.last_log_in = error
  86. save!
  87. end
  88. end
  89. =begin
  90. stream all accounts
  91. Channel.stream
  92. =end
  93. def self.stream
  94. Thread.abort_on_exception = true
  95. auto_reconnect_after = 180
  96. delay_before_reconnect = 70
  97. last_channels = []
  98. loop do
  99. logger.debug { 'stream controll loop' }
  100. current_channels = []
  101. channels = Channel.where('active = ? AND area LIKE ?', true, '%::Account')
  102. channels.each do |channel|
  103. adapter = channel.options[:adapter]
  104. next if adapter.blank?
  105. driver_class = Object.const_get("Channel::Driver::#{adapter.to_classname}")
  106. next if !driver_class.respond_to?(:streamable?)
  107. next if !driver_class.streamable?
  108. channel_id = channel.id.to_s
  109. current_channels.push channel_id
  110. # exit it channel has changed or connection is older then 180 minutes
  111. if @@channel_stream[channel_id].present?
  112. if @@channel_stream[channel_id][:options] != channel.options
  113. logger.info "channel options (#{channel.id}) has changed, stop stream thread"
  114. @@channel_stream[channel_id][:thread].exit
  115. @@channel_stream[channel_id][:thread].join
  116. @@channel_stream[channel_id][:stream_instance].disconnect
  117. @@channel_stream.delete(channel_id)
  118. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  119. next
  120. elsif @@channel_stream[channel_id][:started_at] && @@channel_stream[channel_id][:started_at] < Time.zone.now - auto_reconnect_after.minutes
  121. logger.info "channel (#{channel.id}) reconnect - stream thread is older then #{auto_reconnect_after} minutes, restart thread"
  122. @@channel_stream[channel_id][:thread].exit
  123. @@channel_stream[channel_id][:thread].join
  124. @@channel_stream[channel_id][:stream_instance].disconnect
  125. @@channel_stream.delete(channel_id)
  126. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  127. next
  128. end
  129. end
  130. local_delay_before_reconnect = delay_before_reconnect
  131. if channel.status_in == 'error'
  132. local_delay_before_reconnect = local_delay_before_reconnect * 2
  133. end
  134. if @@channel_stream[channel_id].blank? && @@channel_stream_started_till_at[channel_id].present?
  135. wait_in_seconds = @@channel_stream_started_till_at[channel_id] - (Time.zone.now - local_delay_before_reconnect.seconds)
  136. if wait_in_seconds.positive?
  137. logger.info "skipp channel (#{channel_id}) for streaming, already tried to connect or connection was active within the last #{local_delay_before_reconnect} seconds - wait another #{wait_in_seconds} seconds"
  138. next
  139. end
  140. end
  141. #logger.info "thread stream for channel (#{channel.id}) already running" if @@channel_stream[channel_id].present?
  142. next if @@channel_stream[channel_id].present?
  143. @@channel_stream[channel_id] = {
  144. options: channel.options,
  145. started_at: Time.zone.now,
  146. }
  147. # start channels with delay
  148. sleep @@channel_stream.count
  149. # start threads for each channel
  150. @@channel_stream[channel_id][:thread] = Thread.new do
  151. begin
  152. logger.info "Started stream channel for '#{channel.id}' (#{channel.area})..."
  153. channel.status_in = 'ok'
  154. channel.last_log_in = ''
  155. channel.save!
  156. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  157. @@channel_stream[channel_id] ||= {}
  158. @@channel_stream[channel_id][:stream_instance] = channel.stream_instance
  159. @@channel_stream[channel_id][:stream_instance].stream
  160. @@channel_stream[channel_id][:stream_instance].disconnect
  161. @@channel_stream.delete(channel_id)
  162. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  163. logger.info " ...stopped stream thread for '#{channel.id}'"
  164. rescue => e
  165. error = "Can't use stream for channel (#{channel.id}): #{e.inspect}"
  166. logger.error error
  167. logger.error e.backtrace
  168. channel.status_in = 'error'
  169. channel.last_log_in = error
  170. channel.save!
  171. @@channel_stream.delete(channel_id)
  172. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  173. end
  174. end
  175. end
  176. # cleanup deleted channels
  177. last_channels.each do |channel_id|
  178. next if @@channel_stream[channel_id].blank?
  179. next if current_channels.include?(channel_id)
  180. logger.info "channel (#{channel_id}) not longer active, stop stream thread"
  181. @@channel_stream[channel_id][:thread].exit
  182. @@channel_stream[channel_id][:thread].join
  183. @@channel_stream[channel_id][:stream_instance].disconnect
  184. @@channel_stream.delete(channel_id)
  185. @@channel_stream_started_till_at[channel_id] = Time.zone.now
  186. end
  187. last_channels = current_channels
  188. sleep 20
  189. end
  190. end
  191. =begin
  192. send via account
  193. channel = Channel.where(area: 'Email::Account').first
  194. channel.deliver(mail_params, notification)
  195. =end
  196. def deliver(mail_params, notification = false)
  197. adapter = options[:adapter]
  198. adapter_options = options
  199. if options[:outbound] && options[:outbound][:adapter]
  200. adapter = options[:outbound][:adapter]
  201. adapter_options = options[:outbound][:options]
  202. end
  203. result = nil
  204. begin
  205. # we need to require each channel backend individually otherwise we get a
  206. # 'warning: toplevel constant Twitter referenced by Channel::Driver::Twitter' error e.g.
  207. # so we have to convert the channel name to the filename via Rails String.underscore
  208. # http://stem.ps/rails/2015/01/25/ruby-gotcha-toplevel-constant-referenced-by.html
  209. require "channel/driver/#{adapter.to_filename}"
  210. driver_class = Object.const_get("Channel::Driver::#{adapter.to_classname}")
  211. driver_instance = driver_class.new
  212. result = driver_instance.send(adapter_options, mail_params, notification)
  213. self.status_out = 'ok'
  214. self.last_log_out = ''
  215. save!
  216. rescue => e
  217. error = "Can't use Channel::Driver::#{adapter.to_classname}: #{e.inspect}"
  218. logger.error error
  219. logger.error e.backtrace
  220. self.status_out = 'error'
  221. self.last_log_out = error
  222. save!
  223. raise error
  224. end
  225. result
  226. end
  227. private
  228. def email_address_check
  229. # reset non existing channel_ids
  230. EmailAddress.channel_cleanup
  231. end
  232. end