binlog_event.php

<?php
/* BinlogEvent.php --- 
 * 
 * Filename: BinlogEvent.php
 * Description: 
 * Author: Gu Weigang  * Maintainer: 
 * Created: Fri Nov  8 20:16:59 2013 (+0800)
 * Version: 117777
 * Last-Updated: Tue Nov 19 15:05:08 2013 (+0800)
 *           By: Gu Weigang
 *     Update #: 121
 * 
 */

/* Change Log:
 * 
 * 
 */

/* This program is part of "Baidu Darwin PHP Software"; you can redistribute it and/or
 * modify it under the terms of the Baidu General Private License as
 * published by Baidu Campus.
 * 
 * You should have received a copy of the Baidu General Private License
 * along with this program; see the file COPYING. If not, write to
 * the Baidu Campus NO.10 Shangdi 10th Street Haidian District, Beijing The People's
 * Republic of China, 100085.
 */

/* Code: */

class BinlogEvent
{
    const SUCCESS = 0;
    const FAILURE = 1;
    const IGNORE  = 2;
    const L_EXIT  = 3;
    
    const SIGNAL_QUIT = 100;
        
    public function onUnknownEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onStartEventV3($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onStopEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onIntvarEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onLoadEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onSlaveEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onCreateFileEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onAppendBlockEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onExecLoadEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onDeleteFileEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onNewLoadEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onRandEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onUserVarEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onBeginLoadQueryEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onExecuteLoadQueryEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onTableMapEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onPreGaWriteRowsEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onPreGaUpdateRowsEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onPreGaDeleteRowsEvent($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }

    public function onFormatDescriptionEvent($event, &$message, $params=array())
    {
        return self::IGNORE;                
    }

    public function onQueryEvent($event, &$message, $params=array())
    {
        if($event['query'] == 'BEGIN') {
            $message['position'] = $event['next_position'];
        }
        if($event['query'] == 'COMMIT') {
            $message['position'] = $event['next_position'];
            if(isset($params['controlSignal']) && $params['controlSignal'] === true) {
                return self::L_EXIT;
            }
        }
        return self::IGNORE;
    }
    
    public function onRotateEvent($event, &$message, $params=array())
    {
        $message['filename'] = $event['filename'];
        $message['position'] = $event['position'];        
        return self::IGNORE;
    }

    public function onXidEvent($event, &$message, $params=array())
    {
        $message['position'] = $event['next_position'];
        if(isset($params['controlSignal']) && $params['controlSignal'] === true) {
            return self::L_EXIT;
        }
        return self::IGNORE;
    }

    protected function getHandleClass($dbName, $tblName)
    {
        $dbClass = "\Adpipe\Event\Adapter\\".str_replace("_", "", $dbName);
        $tblClass = ucfirst($tblName);
        $handleClass = $dbClass."\\".$tblClass;
        if(class_exists($handleClass)) {
            $class = new $handleClass();
        } else {
            $class = new \Adpipe\Finance\Logic\Event\Adapter\DefaultEvent();
        }
        return $class;
    }
    
    public function onWriteRowsEvent($event, &$message, $params=array())
    {
        $class = $this->getHandleClass($event['db_name'], $event['table_name']);

        $binlogInfo = array(
            'db'       => $event['db_name'],
            'table'    => $event['table_name'],
            'filename' => $params['filename'],
            'position' => $params['position'],
        );
        
        if(isset($event['rows'])) {
            foreach($event['rows'] as $item) {
                $return = $class->onWrite($item, $entry, $binlogInfo);
                if($return === true) {
                    $message = array_merge($message, $entry);
                }
            }
            if(!empty($message)) return self::SUCCESS;
        }
        return self::IGNORE;
    }

    public function onUpdateRowsEvent($event, &$message, $params=array())
    {
        $class = $this->getHandleClass($event['db_name'], $event['table_name']);

        $binlogInfo = array(
            'db'       => $event['db_name'],
            'table'    => $event['table_name'],
            'filename' => $params['filename'],
            'position' => $params['position'],
        );
        
        if(isset($event['rows'])) {
            foreach($event['rows'] as $_idx => $item) {
                if($_idx % 2 != 0) {
                    // new, old, message entry
                    $return = $class->onUpdate($event['rows'][$_idx-1], $item, $entry, $binlogInfo);
                    // ignore on false, signal-quit on null, public on true
                    if($return === null) {
                        return self::SIGNAL_QUIT;
                    } else if($return === false) {
                        continue;
                    } else if($return === true) {
                        $message = array_merge($message, $entry);
                    }
                }
            }
            if(!empty($message)) return self::SUCCESS;
        }
        return self::IGNORE;
    }

    public function onDeleteRowsEvent($event, &$message, $params=array())
    {
        $class = $this->getHandleClass($event['db_name'], $event['table_name']);

        $binlogInfo = array(
            'db'       => $event['db_name'],
            'table'    => $event['table_name'],
            'filename' => $params['filename'],
            'position' => $params['position'],
        );
        
        if(isset($event['rows'])) {
            foreach($event['rows'] as $item) {
                $return = $class->onDelete($item, $entry, $binlogInfo);
                if($return === true) {
                    $message = array_merge($message, $entry);
                }
            }
            if(!empty($message)) return self::SUCCESS;
        }
        return self::IGNORE;
    }

    public function onIncidentEvent($event, &$message, $params=array())
    {
        return self::IGNORE;
    }

    public function onUserDefined($event, &$message, $params=array())
    {
        return self::IGNORE;        
    }
}

/* BinlogEvent.php ends here */