Class: StreamingImport

Inherits:
Object
  • Object
show all
Includes:
JSONModel
Defined in:
backend/app/lib/streaming_import.rb

Instance Method Summary (collapse)

Methods included from JSONModel

JSONModel, #JSONModel, add_error_handler, all, allow_unmapped_enum_value, backend_url, client_mode?, custom_validations, destroy_model, enum_default_value, enum_values, handle_error, init, load_schema, #models, models, parse_jsonmodel_ref, parse_reference, repository, repository_for, schema_src, set_repository, strict_mode, strict_mode?, with_repository

Constructor Details

- (StreamingImport) initialize(stream, ticker, import_canceled = false, migration = false)

Returns a new instance of StreamingImport

Raises:

  • (StandardError) 


60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'backend/app/lib/streaming_import.rb', line 60

def initialize(stream, ticker, import_canceled = false,  migration = false)
  
  @import_canceled = import_canceled ? import_canceled :  Atomic.new(false)
  @migration = migration ? Atomic.new(true) : Atomic.new(false)
 
  raise StandardError.new("Nothing to stream") unless stream

  @ticker = ticker

  with_status("Reading JSON records") do

    @ticker.tick_estimate = 1000 # this is totally made up, just want to show something

    @tempfile = ASUtils.tempfile('import_stream')

    begin
      while !(buf = stream.read(4096)).nil?
        @tempfile.write(buf)
        ticker.tick
      end
    ensure
      @tempfile.close
    end

  end

  @jstream = StreamingJsonReader.new(@tempfile.path)

  if @jstream.empty?
    @ticker.log("No records were found in the input file!")
  end

  with_status("Validating records and checking links") do
    @logical_urls = load_logical_urls
  end

  with_status("Evaluating record relationships") do
    @dependencies, @position_offsets = load_dependencies
  end

  @limbs_for_reattaching = {}
end

Instance Method Details

- (Object) abort_if_import_canceled 



109
110
111
112
113
114
 
# File 'backend/app/lib/streaming_import.rb', line 109

def abort_if_import_canceled
  if @import_canceled.value
    @ticker.log("Import canceled!")
    raise ImportCanceled.new
  end
end

- (Object) created_records 



104
105
106
 
# File 'backend/app/lib/streaming_import.rb', line 104

def created_records
  @logical_urls.reject {|k, v| v.nil?}
end

- (Object) process 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'backend/app/lib/streaming_import.rb', line 117

def process

  round = 0
  finished = true

  begin
    while true
      round += 1

      finished = true
      progressed = false

      with_status("Saving records: cycle #{round}") do
        @ticker.tick_estimate = @jstream.count
        @jstream.each do |rec|
          abort_if_import_canceled

          uri = rec['uri']
          dependencies = @dependencies[uri]

          if !@logical_urls[uri] && dependencies.all? {|d| @logical_urls[d]}
            # migrate it
            @logical_urls[uri] = do_create(rewrite(rec, @logical_urls))

            progressed = true
          end

          if !@logical_urls[uri]
            finished = false
          end

          @ticker.tick
        end
      end

      if finished
        break
      end

      with_status("Dealing with circular dependencies: cycle #{round}") do
        if !progressed
          run_dependency_breaking_cycle
        end
      end
    end

  ensure
    with_status("Cleaning up") do
      if finished
        reattach_severed_limbs
        touch_toplevel_records
      end

      cleanup
    end
  end

  @logical_urls
end