class GoldenGateEventProducer extends AbstractEventProducer {

  protected void addEventToBuffer(List<TransactionState.PerSourceTransactionalUpdate> dbUpdates, TransactionInfo ti)
      throws DatabusException, UnsupportedKeyException
  {
    if(dbUpdates.size() == 0)
      throw new DatabusException("Cannot handle empty dbUpdates");

    long scn = ti.getScn();
    long timestamp = ti.getTransactionTimeStampNs();
    EventSourceStatistics globalStats = getSource(GLOBAL_SOURCE_ID).getStatisticsBean();

    /**
     * We skip the start scn of the relay, we have already added a EOP for this SCN in the buffer.
     * Why is this not a problem ?
     * There are two cases:
     * 1. When we use the earliest/latest scn if there is no maxScn (We don't really have a start point). So it's really OK to miss the first event.
     * 2. If it's the maxSCN, then event was already seen by the relay.
     */
    if(scn == _startPrevScn.get())
    {
      _log.info("Skipping this transaction, EOP already send for this event");
      return;
    }

    getEventBuffer().startEvents();

    int eventsInTransactionCount = 0;

    List<EventReaderSummary> summaries = new ArrayList<EventReaderSummary>();

    for (int i = 0; i < dbUpdates.size(); ++i)
    {
      GenericRecord record = null;
      TransactionState.PerSourceTransactionalUpdate perSourceUpdate = dbUpdates.get(i);
      short sourceId = (short)perSourceUpdate.getSourceId();
      // prepare stats collection per source
      EventSourceStatistics perSourceStats = getSource(sourceId).getStatisticsBean();

      Iterator<DbUpdateState.DBUpdateImage> dbUpdateIterator = perSourceUpdate.getDbUpdatesSet().iterator();
      int eventsInDbUpdate = 0;
      long dbUpdatesEventsSize = 0;
      long startDbUpdatesMs = System.currentTimeMillis();

      while(dbUpdateIterator.hasNext())  //TODO verify if there is any case where we need to rollback.
      {
        DbUpdateState.DBUpdateImage dbUpdate =  dbUpdateIterator.next();

        //Construct the Databus Event key, determine the key type and construct the key
        Object keyObj = obtainKey(dbUpdate);
        DbusEventKey eventKey = new DbusEventKey(keyObj);

        //Get the logicalparition id
        PartitionFunction partitionFunction = _partitionFunctionHashMap.get((int)sourceId);
        short lPartitionId = partitionFunction.getPartition(eventKey);

        record = dbUpdate.getGenericRecord();
        //Write the event to the buffer
        if (record == null)
          throw new DatabusException("Cannot write event to buffer because record = " + record);

        if(record.getSchema() == null)
          throw new DatabusException("The record does not have a schema (null schema)");

        try
        {
          //Collect stats on number of dbUpdates for one source
          eventsInDbUpdate++;

          //Count of all the events in the current transaction
          eventsInTransactionCount++;
          // Serialize the row
          ByteArrayOutputStream bos = new ByteArrayOutputStream();
          Encoder encoder = new BinaryEncoder(bos);
          GenericDatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>(
              record.getSchema());
          writer.write(record, encoder);
          byte[] serializedValue = bos.toByteArray();

          //Get the md5 for the schema
          SchemaId schemaId = SchemaId.createWithMd5(dbUpdate.getSchema());

          //Determine the operation type and convert to dbus opcode
          DbusOpcode opCode;
          if(dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.INSERT || dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.UPDATE)
          {
            opCode = DbusOpcode.UPSERT;
            if(_log.isDebugEnabled())
              _log.debug("The event with scn "+ scn +" is INSERT/UPDATE");
          }
          else if(dbUpdate.getOpType() == DbUpdateState.DBUpdateImage.OpType.DELETE)
          {
            opCode = DbusOpcode.DELETE;
            if(_log.isDebugEnabled())
              _log.debug("The event with scn "+ scn +" is DELETE");
          }
          else
          {
            throw new DatabusException("Unknown opcode from dbUpdate for event with scn:" + scn);
          }


          //Construct the dbusEvent info
          DbusEventInfo dbusEventInfo = new DbusEventInfo(opCode,
                                                          scn,
                                                          (short)_pConfig.getId(),
                                                          lPartitionId,
                                                          timestamp,
                                                          sourceId,
                                                          schemaId.getByteArray(),
                                                          serializedValue,
                                                          false,
                                                          false);
          dbusEventInfo.setReplicated(dbUpdate.isReplicated());

          perSourceStats.addEventCycle(1, ti.getTransactionTimeRead(), serializedValue.length, scn);
          globalStats.addEventCycle(1, ti.getTransactionTimeRead(), serializedValue.length, scn);

          long tsEnd = System.currentTimeMillis();
          perSourceStats.addTimeOfLastDBAccess(tsEnd);
          globalStats.addTimeOfLastDBAccess(tsEnd);

          //Append to the event buffer
          getEventBuffer().appendEvent(eventKey, dbusEventInfo, _statsCollector);
          _rc.incrementEventCount();
          dbUpdatesEventsSize += serializedValue.length;
        }
        catch (IOException io)
        {
          perSourceStats.addError();
          globalStats.addEmptyEventCycle();
          _log.error("Cannot create byte stream payload: " + dbUpdates.get(i).getSourceId());
        }
      }
      long endDbUpdatesMs = System.currentTimeMillis();
      long dbUpdatesElapsedTimeMs = endDbUpdatesMs - startDbUpdatesMs;

      // Log Event Summary at logical source level
      EventReaderSummary summary = new EventReaderSummary(sourceId, _monitoredSources.get(sourceId).getSourceName(),
                                                          scn, eventsInDbUpdate, dbUpdatesEventsSize,-1L /* Not supported */,
                                                          dbUpdatesElapsedTimeMs,  timestamp, timestamp, -1L /* Not supported */);
      if (_eventsLog.isInfoEnabled())
      {
        _eventsLog.info(summary.toString());
      }
      summaries.add(summary);

      if(_log.isDebugEnabled())
        _log.debug("There are "+ eventsInDbUpdate + " events seen in the current dbUpdate");
    }

    // Log Event Summary at Physical source level
    ReadEventCycleSummary summary = new ReadEventCycleSummary(_pConfig.getName(),
                                                              summaries,
                                                              scn,
                                                              -1 /* Overall time including query time not calculated */);

    if (_eventsLog.isInfoEnabled())
    {
      _eventsLog.info(summary.toString());
    }

    _log.info("Writing "+ eventsInTransactionCount + " events from transaction with scn: " + scn);
    if(scn <= 0)
      throw new DatabusException("Unable to write events to buffer because of negative/zero scn: " + scn);

    getEventBuffer().endEvents(scn, _statsCollector);
    _scn.set(scn);

    if (getMaxScnReaderWriter() != null)
    {
      try
      {
        getMaxScnReaderWriter().saveMaxScn(_scn.get());
      }
      catch (DatabusException e)
      {
        _log.error("Cannot save scn = " + _scn + " for physical source = " + getName(), e);
      }
    }
  }

  public  GoldenGateEventProducer(PhysicalSourceStaticConfig pConfig,
                                 SchemaRegistryService schemaRegistryService,
                                 DbusEventBufferAppendable dbusEventBuffer,
                                 DbusEventsStatisticsCollector statsCollector,
                                 MaxSCNReaderWriter maxScnReaderWriters);

}