src/Subjects/AbstractSubject.php
/**
     * Imports the passed row into the database. If the import failed, the exception
     * will be catched and logged, but the import process will be continued.
     *
     * @param array $row The row with the data to be imported
     *
     * @return void
     */
Imports the passed row into the database. If the import failed, the exception
will be catched and logged, but the import process will be continued.

@param array $row The row with the data to be imported

@return void
Imports the passed row into the database . If the import failed the exception will be catched and logged but the import process will be continued . @param array $row The row with the data to be imported @return void
public function importRow(array $row)
    {
        $this->row = $row;
        $this->lineNumber++;
        $this->skipRow = false;
        $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_ROW_PROCESS_START, $this);
        $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_ROW_PROCESS_START), $this);
        if (sizeof($this->headers) === 0) {
            $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_START, $this);
            $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_START), $this);
            foreach ($this->row as $value => $key) {
                $this->headers[$this->mapAttributeCodeByHeaderMapping($key)] = $value;
            }
            foreach ($this->defaultColumnValues as $name => $value) {
                if (array_key_exists($key = $this->mapAttributeCodeByHeaderMapping($name), $this->headers)) {
                    continue;
                }
                $this->headers[$key] = $columnKey = sizeof($this->headers);
                $this->columnValues[$columnKey] = $value;
            }
            $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_SUCCESS, $this);
            $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_SUCCESS), $this);
        } else {
            $this->row = array_replace($this->row, $this->columnValues);
            $availableObservers = $this->getObservers();
            foreach ($availableObservers as $observers) {
                foreach ($observers as $observer) {
                    if ($this->skipRow) {
                        $this->getSystemLogger()->debug(
                            $this->appendExceptionSuffix(
                                sprintf(
                                    'Skip processing operation "%s" after observer "%s"',
                                    $this->getFullOperationName(),
                                    get_class($observer)
                                )
                            )
                        );
                        break 2;
                    }
                    if ($observer instanceof ObserverInterface) {
                        $this->row = $observer->handle($this);
                    }
                }
            }
        }
        if ($this->lastLog < time() - 59) {
            $this->getSystemLogger()->info(
                sprintf(
                    'Operation "%s" successfully processed "%d (%d)" rows per minute of file "%s"',
                    $this->getFullOperationName(),
                    $this->lineNumber - $this->lastLineNumber,
                    $this->lineNumber,
                    basename($this->getFilename())
                )
            );
            $this->lastLog = time();
            $this->lastLineNumber = $this->lineNumber;
        }
        $this->getSystemLogger()->debug(
            $this->appendExceptionSuffix(
                sprintf(
                    'Successfully processed operation "%s"',
                    implode(' > ', $this->getConfiguration()->getConfiguration()->getOperationNames())
                )
            )
        );
        $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_ROW_PROCESS_SUCCESS, $this);
        $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_ROW_PROCESS_SUCCESS), $this);
    }
public function importRow ( array $ row ) { $ this -> row = $ row ; $ this -> lineNumber ++ ; $ this -> skipRow = false ; $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_ROW_PROCESS_START , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_ROW_PROCESS_START ) , $ this ) ; if ( sizeof ( $ this -> headers ) === 0 ) { $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_START , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_START ) , $ this ) ; foreach ( $ this -> row as $ value => $ key ) { $ this -> headers [ $ this -> mapAttributeCodeByHeaderMapping ( $ key ) ] = $ value ; } foreach ( $ this -> defaultColumnValues as $ name => $ value ) { if ( array_key_exists ( $ key = $ this -> mapAttributeCodeByHeaderMapping ( $ name ) , $ this -> headers ) ) { continue ; } $ this -> headers [ $ key ] = $ columnKey = sizeof ( $ this -> headers ) ; $ this -> columnValues [ $ columnKey ] = $ value ; } $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_SUCCESS , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_HEADER_ROW_PROCESS_SUCCESS ) , $ this ) ; } else { $ this -> row = array_replace ( $ this -> row , $ this -> columnValues ) ; $ availableObservers = $ this -> getObservers ( ) ; foreach ( $ availableObservers as $ observers ) { foreach ( $ observers as $ observer ) { if ( $ this -> skipRow ) { $ this -> getSystemLogger ( ) -> debug ( $ this -> appendExceptionSuffix ( sprintf ( 'Skip processing operation "%s" after observer "%s"' , $ this -> getFullOperationName ( ) , get_class ( $ observer ) ) ) ) ; break 2 ; } if ( $ observer instanceof ObserverInterface ) { $ this -> row = $ observer -> handle ( $ this ) ; } } } } if ( $ this -> lastLog < time ( ) - 59 ) { $ this -> getSystemLogger ( ) -> info ( sprintf ( 'Operation "%s" successfully processed "%d (%d)" rows per minute of file "%s"' , $ this -> getFullOperationName ( ) , $ this -> lineNumber - $ this -> lastLineNumber , $ this -> lineNumber , basename ( $ this -> getFilename ( ) ) ) ) ; $ this -> lastLog = time ( ) ; $ this -> lastLineNumber = $ this -> lineNumber ; } $ this -> getSystemLogger ( ) -> debug ( $ this -> appendExceptionSuffix ( sprintf ( 'Successfully processed operation "%s"' , implode ( ' > ' , $ this -> getConfiguration ( ) -> getConfiguration ( ) -> getOperationNames ( ) ) ) ) ) ; $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_ROW_PROCESS_SUCCESS , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_ROW_PROCESS_SUCCESS ) , $ this ) ; }
Imports the passed row into the database.
Imports the passed row into the database .
// initialize the row // raise the line number and reset the skip row flag // invoke the events that has to be fired before the artfact's row will be processed // initialize the headers with the columns from the first line // invoke the events that has to be fired before the artfact's header row will be processed // iterate over the column name => key an map the header names, if necessary // iterate over the default column values to figure out whether or not the column exists // do nothing, if the column already exists // add the header and the default value for the column // invoke the events that has to be fired when the artfact's header row has been successfully processed // merge the default column value into the actual row // load the available observers // process the observers // invoke the pre-import/import and post-import observers /** @var \TechDivision\Import\Observers\ObserverInterface $observer */ // query whether or not we have to skip the row // log a debug message with the actual line nr/file information // skip the row // if not, set the subject and process the observer // query whether or not a minute has been passed // log the number processed rows per minute // reset the last log time and the line number // log a debug message with the actual line nr/file information // invoke the events that has to be fired when the artfact's row has been successfully processed