public function save($datas, array $identifier = array(), $shardingKey = null, &$affectedRows = null, $replace = false, $rawData = false)
    {
        if ($this->breakChainOnSave) {
            throw new BreakSaveChainException('Source configured to break chain on save');
        }

        //Extract sharding key from identifier
        $originalIdentifier = $identifier;

        //Check if we have to insert or update
        $insert =empty($identifier);

        //find identifiers
        $identifierFieldName = null;
        foreach ($this->configuration['fields'] as $field => $fieldProperties) {
            if (isset($fieldProperties['type']) && 'identifier' == $fieldProperties['type'] && false === array_key_exists($field, $datas)) {
                $identifierFieldName = $field;
                if (array_key_exists($field, $datas) && is_null($datas[$field])) {
                    unset($datas[$field]);
                }
            }
        }

        //Check if we have datas to update or insert
        if (empty($datas)) {
            throw new \Exception("Trying to save empty datas for table {$this->tableName} and identifier ".var_export($identifier, true));
        }

        //Get Connection and table name
        $connection = $this->connectionManager->getConnection(
            $this->schemaManager->getConnectionName($this->tableName, $shardingKey, true),
            true
        );
        $shardingIdentifier = $this->schemaManager->getShardingIdentifier($this->tableName, $shardingKey);
        $tableName = $this->tableName;
        if (isset($shardingIdentifier)) {
            $tableName = sprintf('%1$s_%2$s', $this->tableName, $shardingIdentifier);
        }

        //Parse datas
        if (false === $rawData) {
            $datas = $this->dataParser->encode($datas, $this->configuration['fields']);
        }
        //check if it's an insert or update
        if ($insert) {
            //Insert
            $affectedRows = $connection->insert($tableName, $datas);

            if (isset($identifierFieldName)) {
                return array(
                    'sharding_key' => $shardingKey,
                    $identifierFieldName=>$connection->lastInsertId(),
                );
            } else {
                return $identifier;
            }
        } else {
            if ($replace) {
                //Replace
                $connection->connect();
                $set = array();
                $params = array();
                foreach ($datas as $columnName => $value) {
                    $set[$columnName] = $columnName.' = ?';
                    $params[$columnName] = $value;
                }
                foreach ($identifier as $columnName => $value) {
                    $set[$columnName] = $columnName.' = ?';
                    $params[$columnName] = $value;
                }

                $sql  = 'REPLACE INTO `'.$tableName.'` SET '.implode(', ', $set);

                $affectedRows = $connection->executeUpdate($sql, array_values($params));

                return $originalIdentifier;
            } else {
                //Update
                $affectedRows = $connection->update('`'.$tableName.'`', $datas, $identifier);

                return $originalIdentifier;
            }
        }
    }
public function save($datas, array $identifier = array(), $shardingKey = null, &$affectedRows = null, $replace = false, $rawData = false)
    {
        if ($this->breakChainOnSave) {
            throw new BreakSaveChainException('Source configured to break chain on save');
        }

        //Extract sharding key from identifier
        $originalIdentifier = $identifier;

        //Check if we have to insert or update
        $insert =empty($identifier);

        //find identifiers
        $identifierFieldName = null;
        foreach ($this->configuration['fields'] as $field => $fieldProperties) {
            if (isset($fieldProperties['type']) && 'identifier' == $fieldProperties['type'] && false === array_key_exists($field, $datas)) {
                $identifierFieldName = $field;
                if (array_key_exists($field, $datas) && is_null($datas[$field])) {
                    unset($datas[$field]);
                }
            }
        }

        //Check if we have datas to update or insert
        if (empty($datas)) {
            throw new \Exception("Trying to save empty datas for table {$this->tableName} and identifier ".var_export($identifier, true));
        }

        //Get Connection and table name
        $connection = $this->connectionManager->getConnection(
            $this->schemaManager->getConnectionName($this->tableName, $shardingKey, true),
            true
        );
        $shardingIdentifier = $this->schemaManager->getShardingIdentifier($this->tableName, $shardingKey);
        $tableName = $this->tableName;
        if (isset($shardingIdentifier)) {
            $tableName = sprintf('%1$s_%2$s', $this->tableName, $shardingIdentifier);
        }

        //Parse datas
        if (false === $rawData) {
            $datas = $this->dataParser->encode($datas, $this->configuration['fields']);
        }
        //check if it's an insert or update
        if ($insert) {
            //Insert
            $affectedRows = $connection->insert($tableName, $datas);

            if (isset($identifierFieldName)) {
                return array(
                    'sharding_key' => $shardingKey,
                    $identifierFieldName=>$connection->lastInsertId(),
                );
            } else {
                return $identifier;
            }
        } else {
            if ($replace) {
                //Replace
                $connection->connect();
                $set = array();
                $params = array();
                foreach ($datas as $columnName => $value) {
                    $set[$columnName] = $columnName.' = ?';
                    $params[$columnName] = $value;
                }
                foreach ($identifier as $columnName => $value) {
                    $set[$columnName] = $columnName.' = ?';
                    $params[$columnName] = $value;
                }

                $sql  = 'REPLACE INTO `'.$tableName.'` SET '.implode(', ', $set);

                $affectedRows = $connection->executeUpdate($sql, array_values($params));

                return $originalIdentifier;
            } else {
                //Update
                $affectedRows = $connection->update('`'.$tableName.'`', $datas, $identifier);

                return $originalIdentifier;
            }
        }
    }
@param array $datas
@param array $identifier
@param null  $shardingKey
@param null  $affectedRows
@param bool  $replace
@param bool  $rawData

@return array
@throws BreakSaveChainException
@throws \Doctrine\DBAL\DBALException
@throws \Exception