订阅

package com.zcbus.Service;

import java.util.List;
import java.util.Properties;

import com.zcbus.common.BusClientCmd;
import com.zcbus.common.BusDataVector;
import com.zcbus.common.BusDataVectorColVal;
import com.zcbus.common.BusJsonConverter;
import com.zcbus.common.BusServer;
import com.zcbus.common.ZcbusEnv;
import com.zcbus.common.ZcbusLog;
import com.zcbus.common.ZcbusTable;
import com.zcbus.common.ZcbusTime;

public class ZcbusClientServiceDemo extends BusClientCmd {

    private static int customer_id = 0;
    private static int max_threads = 512;
    private static int logLevel = -1;

    public void close() {

    }

    public void refresh(List<ZcbusTable> tableList) throws Exception {
        if (tableList != null) {
            int i, j, k;
            System.out.println("table count: " + tableList.size());
            for (i = 0; i < tableList.size(); i++) {
                ZcbusTable table = tableList.get(i);
                System.out.println("table name: " + table.getName());
                System.out.println("row count: " + table.getRowCount());
                System.out.println("column count: " + table.getColumnCount());
                for (j = 0; j < table.getColumnCount(); j++) {
                    System.out.print(table.getColumnName(j));
                    System.out.print("\t");
                }
                System.out.println();
                for (j = 0; j < table.getRowCount(); j++) {
                    for (k = 0; k < table.getColumnCount(); k++) {
                        System.out.print(table.getColumnData(j,  k));
                        System.out.print("\t");
                    }
                    System.out.println();
                }
                System.out.println();
            }
        }
    }

    public void processBusdata(byte[] busData, int dataLength) throws Exception {
        BusDataVector bdVector = new BusDataVector();
        int offset = 0;
        while (offset < dataLength) {
            int len = bdVector.deserialize(busData, offset, dataLength);
            if (len <= 0) {
                throw new Exception(String.format("bus data offset %d, length is %d", offset, len));
            }
            /**
             * 自定义数据
             */
            byte bt[] = bdVector.getCustomData();
            if (bt!=null && bt.length > 0) {
                System.err.println("optype :" + bdVector.getOpType());
                System.err.println("trans DATA :" + new String(bt));
                BusJsonConverter busJsonConverter = new BusJsonConverter();
                busJsonConverter.setLoaderTime(ZcbusTime.getCurrentTime());
                busJsonConverter.setSendType("data");
                busJsonConverter.setBatchCode("zcuscode");
                StringBuilder stringBuilder = new StringBuilder();
                busJsonConverter.convetJson(bdVector, stringBuilder);
                System.err.println(stringBuilder.toString());
            }
            List<BusDataVectorColVal> bdvcollist;
            /*
             * 插入、删除
             */
            if (bdVector.getOpType() == BusDataVector.BSDATA_OPTYPE_INSERT
                    || bdVector.getOpType() == BusDataVector.BSDATA_OPTYPE_DELETE) {
                bdvcollist = bdVector.getColList();
                for (int i = 0; i < bdvcollist.size(); i++) {
                    System.err.println(bdvcollist.get(i).getColName() + ":" + bdvcollist.get(i).getColType() + ":"
                            + bdvcollist.get(i).getDataString());
                }
            }
            /*
             * 更新
             */
            if (bdVector.getOpType() == BusDataVector.BSDATA_OPTYPE_UPDATE) {
                bdVector.getOpType();
                // before
                bdvcollist = bdVector.getColBeforeList();
                for (int i = 0; i < bdvcollist.size(); i++) {
                    System.err.println(bdvcollist.get(i).getColName() + ":" + bdvcollist.get(i).getColType() + ":"
                            + bdvcollist.get(i).getDataString());
                }
                // after
                bdvcollist = bdVector.getColList();
                for (int i = 0; i < bdvcollist.size(); i++) {
                    System.err.println(bdvcollist.get(i).getColName() + ":" + bdvcollist.get(i).getColType() + ":"
                            + bdvcollist.get(i).getDataString());
                }
            }
            /*
             * DDL
             */
            if(bdVector.getOpType()==BusDataVector.BSDATA_OPTYPE_DDL) {
                System.err.println(bdVector.getDDlSql());
            }
            offset += len;
        }
        System.out.println(" customerID: " + getCustomerID());
        System.out.println("  controlID: " + getControlID());
        System.out.println("    tableID: " + getTableID());
        System.out.println("     dbName: " + getDbName());
        System.out.println("  tableName: " + getTableName());
        System.out.println("  tableName: " + getObjectID());
        System.out.println("  tableName: " + getThreadId());
        System.out.println("data length: " + dataLength);
    }

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        for (int i = 0; i < args.length; i++) {
            if (0 == args[i].compareTo("-h")) {
                System.out.println(
                        "Usage: \n" + "       -h                 help\n" + "       -log_level 2       log_level\n"
                                + "       -max_thread 128    specify max threads of client service, default 512\n"
                                + "       -customer_id 10058 specify customer id of client service, use for port\n");
                return;
            } else if (0 == args[i].compareTo("-customer_id")) {
                customer_id = Integer.parseInt(args[++i]);
            } else if (0 == args[i].compareTo("-log_level")) {
                logLevel = Integer.parseInt(args[++i]);
            } else if (0 == args[i].compareTo("-max_thread")) {
                max_threads = Integer.parseInt(args[++i]);
            } else {
                ZcbusLog.error("unkown options: " + args[i]);
                return;
            }
        }

        if (customer_id <= 0) {
            ZcbusLog.error("cusomter_id not set");
            return;
        }

        try {
            ZcbusEnv.init();

            String logFile = ZcbusEnv.getZcbusLogPath() + "/log.db_service." + customer_id;
            ZcbusLog.open(logFile);
            ZcbusLog.setLevel(logLevel);

            String workDirectory = ZcbusEnv.getZcbusCachePath() + "/db_service" + customer_id;
            ZcbusEnv.setWorkDirectory(workDirectory);
            ZcbusLog.print(2, "make workDirectory: " + workDirectory);
            ZcbusEnv.createDir(workDirectory);

            if (max_threads < 1)
                max_threads = 512;
            BusServer busServer = new BusServer(customer_id);
            BusServer.registClientCmdClass(ZcbusClientServiceDemo.class);
            busServer.run();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}