Reducing the data persited in the full_data columm

app/models/manageiq/providers/lenovo/physical_infra_manager/event_catcher/stream.rb

@@ -4,6 +4,7 @@ class ManageIQ::Providers::Lenovo::PhysicalInfraManager::EventCatcher::Stream
   def initialize(ems)
     @ems = ems
     @collect_events = true
+    @last_event_ems_ref = last_event_ems_ref(@ems.id)
   end
 
   # Stop capturing events
@@ -15,7 +16,7 @@ def stop
   def each_batch
     $log.info('Starting collect of LXCA events ...')
     while @collect_events
-      yield(events.collect { |e| ManageIQ::Providers::Lenovo::PhysicalInfraManager::EventParser.event_to_hash(e, @ems.id) })
+      yield(parse_events(events))
     end
     $log.info('Stopping collect of LXCA events ...')
   end
@@ -27,12 +28,26 @@ def filter_fields
       { :operation => 'NOT', :field => 'eventClass', :value => '200' },
       { :operation => 'NOT', :field => 'eventClass', :value => '800' }
     ]
-    last_cn_event = get_last_cnn_from_events(@ems.id)
-    cn_operation = { :operation => 'GT', :field => 'cn', :value => last_cn_event.to_s }
-    fields.push(cn_operation) unless last_cn_event.nil?
+
+    unless @last_event_ems_ref.nil?
+      cn_operation = { :operation => 'GT', :field => 'cn', :value => @last_event_ems_ref.to_s }
+      fields.push(cn_operation)
+    end
     fields
   end
 
+  def parse_events(events)
+    return if events.blank?
+
+    parsed_events = events.sort { |x, y| Integer(x.cn) <=> Integer(y.cn) }.collect do |event|
+      ManageIQ::Providers::Lenovo::PhysicalInfraManager::EventParser.event_to_hash(event, @ems.id)
+    end
+
+    # Update the @last_event_ems_ref with the new last ems_ref if to exist new events
+    @last_event_ems_ref = parsed_events.last[:ems_ref]
+    parsed_events
+  end
+
   def events
     expression = { :filterType => 'FIELDNOTREGEXAND', :fields => filter_fields }
 
@@ -54,7 +69,7 @@ def create_event_connection(ems)
                 :port => ems.endpoints.first.port)
   end
 
-  def get_last_cnn_from_events(ems_id)
-    EventStream.where(:ems_id => ems_id).maximum('ems_ref') || 1
+  def last_event_ems_ref(ems_id)
+    EventStream.where(:ems_id => ems_id).maximum('CAST(ems_ref AS int)')
   end
 end

app/models/manageiq/providers/lenovo/physical_infra_manager/event_parser.rb

@@ -1,19 +1,37 @@
 module ManageIQ::Providers::Lenovo::PhysicalInfraManager::EventParser
-  def self.event_to_hash(event, ems_id)
+  def self.event_to_hash(data, ems_id)
+    event = filter_data(data)
     event_hash = {
-      :event_type         => event.msgID,
-      :ems_ref            => event.cn,
-      :source             => "LenovoXclarity",
-      :physical_server_id => get_physical_server_id(event.componentID),
-      :message            => event.msg,
-      :timestamp          => event.timeStamp,
-      :full_data          => event.to_hash,
+      :event_type         => event[:event_type],
+      :ems_ref            => event[:ems_ref],
+      :source             => event[:source],
+      :physical_server_id => get_physical_server_id(event[:component_id]),
+      :message            => event[:message],
+      :timestamp          => event[:timestamp],
+      :full_data          => event,
       :ems_id             => ems_id
     }
-
     event_hash
   end
 
+  def self.filter_data(data)
+    {
+      :component_id   => data.componentID,
+      :component_type => data.typeText,
+      :event_type     => data.msgID,
+      :ems_ref        => data.cn,
+      :message        => data.msg,
+      :parent_uuid    => data.senderUUID,
+      :parent_name    => data.systemName,
+      :parent_model   => data.systemTypeModelText,
+      :parent_type    => data.systemTypeText,
+      :severity_id    => data.severity,
+      :severity       => data.severityText,
+      :source         => 'LenovoXclarity',
+      :timestamp      => data.timeStamp,
+    }
+  end
+
   def self.get_physical_server_id(ems_ref)
     PhysicalServer.find_by(:ems_ref => ems_ref).try(:id)
   end

spec/models/manageiq/providers/lenovo/physical_infra_manager/event_catcher/stream_spec.rb

@@ -16,13 +16,20 @@
     it "yields a valid event" do
       VCR.use_cassette(described_class.name.underscore.to_s) do
         result = []
+        pool = 2
         stream.each_batch do |events|
-          result = events
-          stream.stop
-          expect($log).to receive(:info).with(/Stopping collect of LXCA events .../)
+          result.push(*events)
+          pool -= 1
+          if pool < 1
+            stream.stop
+            expect(events.count).to be == 1
+            expect($log).to receive(:info).with(/Stopping collect of LXCA events .../)
+          else
+            expect(events.count).to be == 20
+          end
         end
-        expect(result.count).to be == 20
-        expect(result.all? { |item| item[:full_data]['eventClass'] == 400 }).to be true
+        expect(result.count).to be == 21
+        expect(result.all? { |item| item[:full_data][:severity_id] == 200 }).to be true
       end
     end
   end