method sync documentation in bbn\Db\Sync
function(bbn\Db $db, $dbs = '', $sync_table = '', $num_try = 0)
{
if (!self::isInit()) {
die("DB sync is not initiated");
}
self::disable();
$mode_db = self::$current_connection->getErrorMode();
$mode_dbs = self::$sync_connection->getErrorMode();
self::$current_connection->setErrorMode("continue");
self::$sync_connection->setErrorMode("continue");
$num_try++;
$to_log = [
'deleted_sync' => 0,
'deleted_real' => 0,
'updated_sync' => 0,
'updated_real' => 0,
'inserted_sync' => 0,
'inserted_real' => 0,
'num_problems' => 0,
'problems' => []
];
$to_log['deleted_sync'] = self::deleteCompleted();
$retry = false;
// Selecting the entries inserted
$ds = self::$sync_connection->rselectAll(
self::$sync_table, ['id', 'tab', 'vals', 'chrono'], [
['db', '!=', self::$current_connection->getCurrent()],
['state', '=', 0],
['action', 'LIKE', 'INSERT']
], [
'chrono' => 'ASC',
'id' => 'ASC'
]
);
// They just have to be inserted
foreach ($ds as $i => $d){
if (isset(self::$methods['cbf1'])) {
self::cbf1($d);
}
$vals = X::jsonBase64Decode($d['vals']);
if (!\is_array($vals)) {
$to_log['num_problems']++;
$to_log['problems'][] = "Hey, look urgently at the row $d[id]!";
}
elseif (self::$current_connection->insert($d['tab'], $vals)) {
if (isset(self::$methods['cbf2'])) {
self::cbf2($d);
}
$to_log['inserted_sync']++;
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
}
elseif (self::$current_connection->select($d['tab'], [], $vals)) {
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
}
else{
if ($num_try > self::$max_retry) {
$to_log['num_problems']++;
$to_log['problems'][] = "Problem while syncing (insert), check data with status 5 and ID ".$d['id'];
self::$sync_connection->update(self::$sync_table, ["state" => 5], ["id" => $d['id']]);
}
$retry = 1;
}
}
// Selecting the entries modified and deleted in the twin DB,
// ordered by table and rows (so the same go together)
$ds = self::$sync_connection->rselectAll(
self::$sync_table, ['id', 'tab', 'action', 'rows', 'vals', 'chrono'], [
['db', '!=', self::$current_connection->getCurrent()],
['state', '=', 0],
['rows', '!=', '[]'],
['action', '!=', 'insert']
], [
'tab' => 'ASC',
'rows' => 'ASC',
'chrono' => 'ASC',
'id' => 'ASC'
]
);
foreach ($ds as $i => $d){
// Executing the first callback
$d['rows'] = X::jsonBase64Decode($d['rows']);
$d['vals'] = X::jsonBase64Decode($d['vals']);
if (isset(self::$methods['cbf1'])) {
self::cbf1($d);
}
// Proceeding to the actions: delete is before
if (strtolower($d['action']) === 'delete') {
if (self::$current_connection->delete($d['tab'], $d['rows'])) {
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
$to_log['deleted_real']++;
}
elseif (!self::$current_connection->select($d['tab'], [], $d['rows'])) {
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
}
else{
if ($num_try > self::$max_retry) {
self::$sync_connection->update(self::$sync_table, ["state" => 5], ["id" => $d['id']]);
$to_log['num_problems']++;
$to_log['problems'][] = "Problem while syncing (delete), check data with status 5 and ID ".$d['id'];
}
$retry = 1;
}
}
// Checking if there is another change done to this record and when in the twin DB
$next_time = (
isset($ds[$i+1]) &&
($ds[$i+1]['tab'] === $d['tab']) &&
($ds[$i+1]['rows'] === $d['rows'])
) ? $ds[$i+1]['chrono'] : microtime();
// Looking for the actions done on this specific record in our database
// between the twin change and the next (or now if there is no other change)
$each = self::$sync_connection->rselectAll(
self::$sync_table, ['id', 'chrono', 'action', 'vals'], [
['db', '=', self::$current_connection->getCurrent()],
['tab', '=', $d['tab']],
['rows', '=', $d['rows']],
['chrono', '>=', $d['chrono']],
['chrono', '<', $next_time],
]
);
if (\count($each) > 0) {
$to_log['num_problems']++;
$to_log['problems'][] = "Conflict!";
$to_log['problems'][] = $d;
foreach ($each as $e){
$e['vals'] = X::jsonBase64Decode($e['vals']);
// If it's deleted locally and updated on the twin we restore
if (strtolower($e['action']) === 'delete') {
if (strtolower($d['action']) === 'update') {
if (!self::$current_connection->insertUpdate(
$d['tab'],
X::mergeArrays(
$e['vals'],
$d['vals']
)
)
) {
$to_log['num_problems']++;
$to_log['problems'][] = "insert_update number 1 had a problem";
}
}
}
// If it's updated locally and deleted in the twin we restore
elseif (strtolower($e['action']) === 'update') {
if (strtolower($d['action']) === 'delete') {
if (!self::$current_connection->insertUpdate($d['tab'], X::mergeArrays($d['vals'], $e['vals']))) {
$to_log['num_problems']++;
$to_log['problems'][] = "insert_update had a problem";
}
}
// If it's updated locally and in the twin we merge the values for the update
elseif (strtolower($d['action']) === 'update') {
$d['vals'] = X::mergeArrays($d['vals'], $e['vals']);
}
}
}
}
// Proceeding to the actions update is after in case we needed to restore
if (strtolower($d['action']) === 'update') {
X::log(X::mergeArrays($d['rows'], $d['vals']), 'synct');
if (self::$current_connection->update($d['tab'], $d['vals'], $d['rows'])) {
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
$to_log['updated_real']++;
}
elseif (self::$current_connection->count($d['tab'], X::mergeArrays($d['rows'], $d['vals']))) {
self::$sync_connection->update(self::$sync_table, ["state" => 1], ["id" => $d['id']]);
}
else{
if ($num_try > self::$max_retry) {
self::$sync_connection->update(self::$sync_table, ["state" => 5], ["id" => $d['id']]);
$to_log['num_problems']++;
$to_log['problems'][] = "Problem while syncing (update), check data with status 5 and ID ".$d['id'];
}
$retry = 1;
}
}
// Callback number 2
if (isset(self::$methods['cbf2'])) {
self::cbf2($d);
}
}
$res = [];
foreach ($to_log as $k => $v){
if (!empty($v)) {
$res[$k] = $v;
}
}
self::$current_connection->setErrorMode($mode_db);
self::$sync_connection->setErrorMode($mode_dbs);
self::enable();
if ($retry && ( $num_try <= self::$max_retry )) {
$res = X::mergeArrays($res, self::sync($db, $dbs, $sync_table, $num_try));
}
return $res;
}
BBN is a suite of PHP and JS libraries and VueJS components - all open-source! bbn.io, build applications, the quick way
This website uses cookies to ensure you get the best experience on our website.
© 2011-2024
BBN Solutions
