Class: Taps::Operation
- Inherits:
-
Object
- Object
- Taps::Operation
- Defined in:
- lib/taps/operation.rb
Instance Attribute Summary collapse
-
#database_url ⇒ Object
readonly
Returns the value of attribute database_url.
-
#opts ⇒ Object
readonly
Returns the value of attribute opts.
-
#remote_url ⇒ Object
readonly
Returns the value of attribute remote_url.
-
#session_uri ⇒ Object
readonly
Returns the value of attribute session_uri.
Class Method Summary collapse
Instance Method Summary collapse
- #apply_table_filter(tables) ⇒ Object
- #catch_errors ⇒ Object
- #close_session ⇒ Object
- #completed_tables ⇒ Object
- #compression_disabled? ⇒ Boolean
- #db ⇒ Object
- #default_chunksize ⇒ Object
- #exclude_tables ⇒ Object
- #exiting? ⇒ Boolean
- #file_prefix ⇒ Object
- #format_number(num) ⇒ Object
- #http_headers(extra = {}) ⇒ Object
- #indexes_first? ⇒ Boolean
-
#initialize(database_url, remote_url, opts = {}) ⇒ Operation
constructor
A new instance of Operation.
- #log ⇒ Object
- #resuming? ⇒ Boolean
- #safe_database_url ⇒ Object
- #safe_remote_url ⇒ Object
- #safe_url(url) ⇒ Object
- #server ⇒ Object
- #session_resource ⇒ Object
- #set_session(uri) ⇒ Object
- #setup_signal_trap ⇒ Object
- #skip_schema? ⇒ Boolean
- #store_session ⇒ Object
- #stream_state ⇒ Object
- #stream_state=(val) ⇒ Object
- #table_filter ⇒ Object
- #to_hash ⇒ Object
- #verify_server ⇒ Object
Constructor Details
#initialize(database_url, remote_url, opts = {}) ⇒ Operation
Returns a new instance of Operation.
19 20 21 22 23 24 25 |
# File 'lib/taps/operation.rb', line 19 def initialize(database_url, remote_url, opts = {}) @database_url = database_url @remote_url = remote_url @opts = opts @exiting = false @session_uri = opts[:session_uri] end |
Instance Attribute Details
#database_url ⇒ Object (readonly)
Returns the value of attribute database_url.
16 17 18 |
# File 'lib/taps/operation.rb', line 16 def database_url @database_url end |
#opts ⇒ Object (readonly)
Returns the value of attribute opts.
16 17 18 |
# File 'lib/taps/operation.rb', line 16 def opts @opts end |
#remote_url ⇒ Object (readonly)
Returns the value of attribute remote_url.
16 17 18 |
# File 'lib/taps/operation.rb', line 16 def remote_url @remote_url end |
#session_uri ⇒ Object (readonly)
Returns the value of attribute session_uri.
17 18 19 |
# File 'lib/taps/operation.rb', line 17 def session_uri @session_uri end |
Class Method Details
.factory(type, database_url, remote_url, opts) ⇒ Object
220 221 222 223 224 225 226 227 228 229 230 |
# File 'lib/taps/operation.rb', line 220 def self.factory(type, database_url, remote_url, opts) type = :resume if opts[:resume] klass = case type when :pull then Taps::Pull when :push then Taps::Push when :resume then eval(opts[:klass]) else raise "Unknown Operation Type -> #{type}" end klass.new(database_url, remote_url, opts) end |
Instance Method Details
#apply_table_filter(tables) ⇒ Object
47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
# File 'lib/taps/operation.rb', line 47 def apply_table_filter(tables) return tables unless table_filter || exclude_tables re = table_filter ? Regexp.new(table_filter) : nil if tables.is_a?(Hash) ntables = {} tables.each do |t, d| if !exclude_tables.include?(t.to_s) && (!re || !re.match(t.to_s).nil?) ntables[t] = d end end ntables else tables.reject { |t| exclude_tables.include?(t.to_s) || (re && re.match(t.to_s).nil?) } end end |
#catch_errors ⇒ Object
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 |
# File 'lib/taps/operation.rb', line 196 def catch_errors verify_server begin yield close_session rescue RestClient::Exception, Taps::BaseError => e store_session if e.is_a?(Taps::BaseError) puts '!!! Caught Server Exception' puts "#{e.class}: #{e.}" puts "\n#{e.original_backtrace}" if e.original_backtrace exit(1) elsif e.respond_to?(:response) puts '!!! Caught Server Exception' puts "HTTP CODE: #{e.http_code}" puts e.response.to_s exit(1) else raise end end end |
#close_session ⇒ Object
148 149 150 |
# File 'lib/taps/operation.rb', line 148 def close_session @session_resource.delete(http_headers) if @session_resource end |
#completed_tables ⇒ Object
112 113 114 |
# File 'lib/taps/operation.rb', line 112 def completed_tables opts[:completed_tables] ||= [] end |
#compression_disabled? ⇒ Boolean
124 125 126 |
# File 'lib/taps/operation.rb', line 124 def compression_disabled? !!opts[:disable_compression] end |
#db ⇒ Object
128 129 130 |
# File 'lib/taps/operation.rb', line 128 def db @db ||= Sequel.connect(database_url) end |
#default_chunksize ⇒ Object
108 109 110 |
# File 'lib/taps/operation.rb', line 108 def default_chunksize opts[:default_chunksize] end |
#exclude_tables ⇒ Object
43 44 45 |
# File 'lib/taps/operation.rb', line 43 def exclude_tables opts[:exclude_tables] || [] end |
#exiting? ⇒ Boolean
88 89 90 |
# File 'lib/taps/operation.rb', line 88 def exiting? !!@exiting end |
#file_prefix ⇒ Object
27 28 29 |
# File 'lib/taps/operation.rb', line 27 def file_prefix 'op' end |
#format_number(num) ⇒ Object
174 175 176 |
# File 'lib/taps/operation.rb', line 174 def format_number(num) num.to_s.gsub(/(\d)(?=(\d\d\d)+(?!\d))/, '\\1,') end |
#http_headers(extra = {}) ⇒ Object
164 165 166 167 168 169 170 171 172 |
# File 'lib/taps/operation.rb', line 164 def http_headers(extra = {}) base = { taps_version: Taps::Version.current } base[:accept_encoding] = if compression_disabled? '' else 'gzip, deflate' end base.merge(extra) end |
#indexes_first? ⇒ Boolean
35 36 37 |
# File 'lib/taps/operation.rb', line 35 def indexes_first? !!opts[:indexes_first] end |
#resuming? ⇒ Boolean
104 105 106 |
# File 'lib/taps/operation.rb', line 104 def resuming? opts[:resume] == true end |
#safe_database_url ⇒ Object
160 161 162 |
# File 'lib/taps/operation.rb', line 160 def safe_database_url safe_url(database_url) end |
#safe_remote_url ⇒ Object
156 157 158 |
# File 'lib/taps/operation.rb', line 156 def safe_remote_url safe_url(remote_url) end |
#safe_url(url) ⇒ Object
152 153 154 |
# File 'lib/taps/operation.rb', line 152 def safe_url(url) url.sub(/\/\/(.+?)?:(.*?)@/, '//\1:[hidden]@') end |
#server ⇒ Object
132 133 134 |
# File 'lib/taps/operation.rb', line 132 def server @server ||= RestClient::Resource.new(remote_url) end |
#session_resource ⇒ Object
136 137 138 139 140 141 |
# File 'lib/taps/operation.rb', line 136 def session_resource @session_resource ||= begin @session_uri ||= server['sessions'].post('', http_headers).to_s server[@session_uri] end end |
#set_session(uri) ⇒ Object
143 144 145 146 |
# File 'lib/taps/operation.rb', line 143 def set_session(uri) session_uri = uri @session_resource = server[session_uri] end |
#setup_signal_trap ⇒ Object
92 93 94 95 96 97 98 99 100 101 102 |
# File 'lib/taps/operation.rb', line 92 def setup_signal_trap trap('INT') do puts "\nCompleting current action..." @exiting = true end trap('TERM') do puts "\nCompleting current action..." @exiting = true end end |
#skip_schema? ⇒ Boolean
31 32 33 |
# File 'lib/taps/operation.rb', line 31 def skip_schema? !!opts[:skip_schema] end |
#store_session ⇒ Object
68 69 70 71 72 73 74 |
# File 'lib/taps/operation.rb', line 68 def store_session file = "#{file_prefix}_#{Time.now.strftime('%Y%m%d%H%M')}.dat" puts "\nSaving session to #{file}.." File.open(file, 'w') do |f| f.write(::OkJson.encode(to_hash)) end end |
#stream_state ⇒ Object
116 117 118 |
# File 'lib/taps/operation.rb', line 116 def stream_state opts[:stream_state] ||= {} end |
#stream_state=(val) ⇒ Object
120 121 122 |
# File 'lib/taps/operation.rb', line 120 def stream_state=(val) opts[:stream_state] = val end |
#table_filter ⇒ Object
39 40 41 |
# File 'lib/taps/operation.rb', line 39 def table_filter opts[:table_filter] end |
#to_hash ⇒ Object
76 77 78 79 80 81 82 83 84 85 86 |
# File 'lib/taps/operation.rb', line 76 def to_hash { klass: self.class.to_s, database_url: database_url, remote_url: remote_url, session_uri: session_uri, stream_state: stream_state, completed_tables: completed_tables, table_filter: table_filter } end |
#verify_server ⇒ Object
178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 |
# File 'lib/taps/operation.rb', line 178 def verify_server server['/'].get(http_headers) rescue RestClient::RequestFailed => e if e.http_code == 417 puts "#{safe_remote_url} is running a different minor version of taps." puts e.response.to_s exit(1) else raise end rescue RestClient::Unauthorized puts "Bad credentials given for #{safe_remote_url}" exit(1) rescue Errno::ECONNREFUSED puts "Can't connect to #{safe_remote_url}. Please check that it's running" exit(1) end |