src/Subjects/AbstractSubject.php
/**
     * Imports the content of the file with the passed filename.
     *
     *
     * @param string $serial   The serial of the actual import
     * @param string $filename The filename to process
     *
     * @return void
     * @throws \Exception Is thrown, if the import can't be processed
     */
Imports the content of the file with the passed filename.
@param string $serial   The serial of the actual import
@param string $filename The filename to process

@return void
@throws \Exception Is thrown, if the import can't be processed
Imports the content of the file with the passed filename . @param string $serial The serial of the actual import @param string $filename The filename to process @return void @throws \ Exception Is thrown if the import can ' t be processed
public function import($serial, $filename)
    {
        try {
            $this->setSerial($serial);
            $this->setFilename($filename);
            $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_PROCESS_START, $this);
            $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_PROCESS_START), $this);
            $systemLogger = $this->getSystemLogger();
            $inProgressFilename = sprintf('%s.inProgress', $filename);
            $importedFilename = sprintf('%s.imported', $filename);
            $failedFilename = sprintf('%s.failed', $filename);
            if ($this->isFile($failedFilename) ||
                $this->isFile($importedFilename) ||
                $this->isFile($inProgressFilename)
            ) {
                $systemLogger->debug(sprintf('Import running, found inProgress file "%s"', $inProgressFilename));
                return;
            }
            $this->touch($inProgressFilename);
            $startTime = microtime(true);
            $this->lastLog = time();
            $systemLogger->info(
                sprintf('Now start processing file "%s"', basename($filename)),
                array('operation-name' => $operationName = $this->getFullOperationName())
            );
            $this->getImportAdapter()->import(array($this, 'importRow'), $filename);
            $endTime = microtime(true) - $startTime;
            $systemLogger->log(
                $skippedRows = $this->getSkippedRows() > 0 ? LogLevel::WARNING : LogLevel::NOTICE,
                sprintf(
                    'Successfully processed file "%s" with "%d" lines (skipping "%d") in "%f" s',
                    basename($filename),
                    $this->getLineNumber() - 1,
                    $skippedRows,
                    $endTime
                ),
                array('operation-name' => $operationName)
            );
            if ($this->getConfiguration()->isCreatingImportedFile()) {
                $this->rename($inProgressFilename, $importedFilename);
            } else {
                $this->getFilesystemAdapter()->delete($inProgressFilename);
            }
            $this->mergeStatus(
                array(
                    RegistryKeys::STATUS => array(
                        RegistryKeys::FILES => array(
                            $filename => array(
                                $this->getUniqueId() => array(
                                    RegistryKeys::STATUS => 1,
                                    RegistryKeys::SKIPPED_ROWS => $this->getSkippedRows(),
                                    RegistryKeys::PROCESSED_ROWS => $this->getLineNumber() - 1
                                )
                            )
                        )
                    )
                )
            );
            $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_PROCESS_SUCCESS, $this);
            $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_PROCESS_SUCCESS), $this);
        } catch (\Exception $e) {
            $this->rename($inProgressFilename, $failedFilename);
            $this->write($failedFilename, $e->__toString());
             with the error message
            $this->mergeStatus(
                array(
                    RegistryKeys::STATUS => array(
                        RegistryKeys::FILES => array(
                            $filename => array(
                                $this->getUniqueId() => array(
                                    RegistryKeys::STATUS         => 2,
                                    RegistryKeys::ERROR_MESSAGE  => $e->getMessage(),
                                    RegistryKeys::SKIPPED_ROWS => $this->getSkippedRows(),
                                    RegistryKeys::PROCESSED_ROWS => $this->getLineNumber() - 1
                                )
                            )
                        )
                    )
                )
            );
            $this->getEmitter()->emit(EventNames::SUBJECT_ARTEFACT_PROCESS_FAILURE, $this, $e);
            $this->getEmitter()->emit($this->getEventName(EventNames::SUBJECT_ARTEFACT_PROCESS_FAILURE), $this, $e);
            if ($e instanceof WrappedColumnException) {
                throw $e;
            }
            throw $this->wrapException(array(), $e);
        }
    }
public function import ( $ serial , $ filename ) { try { $ this -> setSerial ( $ serial ) ; $ this -> setFilename ( $ filename ) ; $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_PROCESS_START , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_PROCESS_START ) , $ this ) ; $ systemLogger = $ this -> getSystemLogger ( ) ; $ inProgressFilename = sprintf ( '%s.inProgress' , $ filename ) ; $ importedFilename = sprintf ( '%s.imported' , $ filename ) ; $ failedFilename = sprintf ( '%s.failed' , $ filename ) ; if ( $ this -> isFile ( $ failedFilename ) || $ this -> isFile ( $ importedFilename ) || $ this -> isFile ( $ inProgressFilename ) ) { $ systemLogger -> debug ( sprintf ( 'Import running, found inProgress file "%s"' , $ inProgressFilename ) ) ; return ; } $ this -> touch ( $ inProgressFilename ) ; $ startTime = microtime ( true ) ; $ this -> lastLog = time ( ) ; $ systemLogger -> info ( sprintf ( 'Now start processing file "%s"' , basename ( $ filename ) ) , array ( 'operation-name' => $ operationName = $ this -> getFullOperationName ( ) ) ) ; $ this -> getImportAdapter ( ) -> import ( array ( $ this , 'importRow' ) , $ filename ) ; $ endTime = microtime ( true ) - $ startTime ; $ systemLogger -> log ( $ skippedRows = $ this -> getSkippedRows ( ) > 0 ? LogLevel :: WARNING : LogLevel :: NOTICE , sprintf ( 'Successfully processed file "%s" with "%d" lines (skipping "%d") in "%f" s' , basename ( $ filename ) , $ this -> getLineNumber ( ) - 1 , $ skippedRows , $ endTime ) , array ( 'operation-name' => $ operationName ) ) ; if ( $ this -> getConfiguration ( ) -> isCreatingImportedFile ( ) ) { $ this -> rename ( $ inProgressFilename , $ importedFilename ) ; } else { $ this -> getFilesystemAdapter ( ) -> delete ( $ inProgressFilename ) ; } $ this -> mergeStatus ( array ( RegistryKeys :: STATUS => array ( RegistryKeys :: FILES => array ( $ filename => array ( $ this -> getUniqueId ( ) => array ( RegistryKeys :: STATUS => 1 , RegistryKeys :: SKIPPED_ROWS => $ this -> getSkippedRows ( ) , RegistryKeys :: PROCESSED_ROWS => $ this -> getLineNumber ( ) - 1 ) ) ) ) ) ) ; $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_PROCESS_SUCCESS , $ this ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_PROCESS_SUCCESS ) , $ this ) ; } catch ( \ Exception $ e ) { $ this -> rename ( $ inProgressFilename , $ failedFilename ) ; $ this -> write ( $ failedFilename , $ e -> __toString ( ) ) ; $ this -> mergeStatus ( array ( RegistryKeys :: STATUS => array ( RegistryKeys :: FILES => array ( $ filename => array ( $ this -> getUniqueId ( ) => array ( RegistryKeys :: STATUS => 2 , RegistryKeys :: ERROR_MESSAGE => $ e -> getMessage ( ) , RegistryKeys :: SKIPPED_ROWS => $ this -> getSkippedRows ( ) , RegistryKeys :: PROCESSED_ROWS => $ this -> getLineNumber ( ) - 1 ) ) ) ) ) ) ; $ this -> getEmitter ( ) -> emit ( EventNames :: SUBJECT_ARTEFACT_PROCESS_FAILURE , $ this , $ e ) ; $ this -> getEmitter ( ) -> emit ( $ this -> getEventName ( EventNames :: SUBJECT_ARTEFACT_PROCESS_FAILURE ) , $ this , $ e ) ; if ( $ e instanceof WrappedColumnException ) { throw $ e ; } throw $ this -> wrapException ( array ( ) , $ e ) ; } }
Imports the content of the file with the passed filename.
Imports the content of the file with the passed filename .
// initialize the serial/filename // invoke the events that has to be fired before the artfact will be processed // load the system logger instance // prepare the flag filenames // query whether or not the file has already been imported // log a debug message and exit // flag file as in progress // track the start time // initialize the last time we've logged the counter with the processed rows per minute // log a message that the file has to be imported // let the adapter process the file // track the time needed for the import in seconds // log a message that the file has successfully been imported, // use log level warning ONLY if rows have been skipped // rename flag file, because import has been successfull // update the status // invoke the events that has to be fired when the artfact has been successfully processed // rename the flag file, because import failed and write the stack trace // update the status with the error message // invoke the events that has to be fired when the artfact can't be processed // do not wrap the exception if not already done // else wrap and throw the exception