channel.rb 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263
  1. # Copyright (C) 2012-2016 Zammad Foundation, http://zammad-foundation.org/
  2. class Channel < ApplicationModel
  3. load 'channel/assets.rb'
  4. include Channel::Assets
  5. belongs_to :group, class_name: 'Group'
  6. store :options
  7. store :preferences
  8. after_create :email_address_check
  9. after_update :email_address_check
  10. after_destroy :email_address_check
  11. # rubocop:disable Style/ClassVars
  12. @@channel_stream = {}
  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. last_channels = []
  96. loop do
  97. logger.debug 'stream controll loop'
  98. current_channels = []
  99. channels = Channel.where('active = ? AND area LIKE ?', true, '%::Account')
  100. channels.each { |channel|
  101. next if channel.options[:adapter] != 'twitter'
  102. current_channels.push channel.id
  103. # exit it channel has changed
  104. if @@channel_stream[channel.id] && @@channel_stream[channel.id][:updated_at] != channel.updated_at
  105. logger.debug "channel (#{channel.id}) has changed, restart thread"
  106. @@channel_stream[channel.id][:thread].exit
  107. @@channel_stream[channel.id][:thread].join
  108. @@channel_stream[channel.id][:stream_instance].disconnect
  109. @@channel_stream[channel.id] = false
  110. end
  111. #logger.debug "thread for channel (#{channel.id}) already running" if @@channel_stream[channel.id]
  112. next if @@channel_stream[channel.id]
  113. @@channel_stream[channel.id] = {
  114. updated_at: channel.updated_at
  115. }
  116. # start channels with delay
  117. sleep @@channel_stream.count
  118. # start threads for each channel
  119. @@channel_stream[channel.id][:thread] = Thread.new {
  120. begin
  121. logger.info "Started stream channel for '#{channel.id}' (#{channel.area})..."
  122. @@channel_stream[channel.id][:stream_instance] = channel.stream_instance
  123. @@channel_stream[channel.id][:stream_instance].stream
  124. @@channel_stream[channel.id][:stream_instance].disconnect
  125. @@channel_stream[channel.id] = false
  126. logger.debug " ...stopped thread for '#{channel.id}'"
  127. rescue => e
  128. error = "Can't use channel (#{channel.id}): #{e.inspect}"
  129. logger.error error
  130. logger.error e.backtrace
  131. channel.status_in = 'error'
  132. channel.last_log_in = error
  133. channel.save
  134. @@channel_stream[channel.id] = false
  135. end
  136. }
  137. }
  138. # cleanup deleted channels
  139. last_channels.each { |channel_id|
  140. next if !@@channel_stream[channel_id]
  141. next if current_channels.include?(channel_id)
  142. logger.debug "channel (#{channel_id}) not longer active, stop thread"
  143. @@channel_stream[channel_id][:thread].exit
  144. @@channel_stream[channel_id][:thread].join
  145. @@channel_stream[channel_id][:stream_instance].disconnect
  146. @@channel_stream[channel_id] = false
  147. }
  148. last_channels = current_channels
  149. sleep 30
  150. end
  151. end
  152. =begin
  153. send via account
  154. channel = Channel.where(area: 'Email::Account').first
  155. channel.deliver(mail_params, notification)
  156. =end
  157. def deliver(mail_params, notification = false)
  158. # ignore notifications in developer mode
  159. if notification == true && Setting.get('developer_mode') == true
  160. logger.info "Do not send notification #{mail_params.inspect} because of enabled developer_mode"
  161. return
  162. end
  163. adapter = options[:adapter]
  164. adapter_options = options
  165. if options[:outbound] && options[:outbound][:adapter]
  166. adapter = options[:outbound][:adapter]
  167. adapter_options = options[:outbound][:options]
  168. end
  169. result = nil
  170. begin
  171. # we need to require each channel backend individually otherwise we get a
  172. # 'warning: toplevel constant Twitter referenced by Channel::Driver::Twitter' error e.g.
  173. # so we have to convert the channel name to the filename via Rails String.underscore
  174. # http://stem.ps/rails/2015/01/25/ruby-gotcha-toplevel-constant-referenced-by.html
  175. require "channel/driver/#{adapter.to_filename}"
  176. driver_class = Object.const_get("Channel::Driver::#{adapter.to_classname}")
  177. driver_instance = driver_class.new
  178. result = driver_instance.send(adapter_options, mail_params, notification)
  179. self.status_out = 'ok'
  180. self.last_log_out = ''
  181. save
  182. rescue => e
  183. error = "Can't use Channel::Driver::#{adapter.to_classname}: #{e.inspect}"
  184. logger.error error
  185. logger.error e.backtrace
  186. self.status_out = 'error'
  187. self.last_log_out = error
  188. save
  189. raise error
  190. end
  191. result
  192. end
  193. private
  194. def email_address_check
  195. # reset non existing channel_ids
  196. EmailAddress.channel_cleanup
  197. end
  198. end