public class DOWithMetadata extends DigitalObject{

    public static final String METADATA = "metadata";

    public static DOWithMetadata fromDO(DigitalObject digitalObject){
        if(digitalObject.attributes == null || digitalObject.attributes.get(METADATA) == null){
            return null;
        }
        return (DOWithMetadata)digitalObject;
    }

    public DOWithMetadata(String id, DoType type) {
        super(id, type);
    }

    public void addMetadata(String key, String value){
        if(attributes==null || attributes.get(METADATA) == null){
            addAttribute(METADATA, new JsonObject());
        }
        attributes.get(METADATA).getAsJsonObject().addProperty(key,value);
    }

    public JsonObject getMetadata(){
        if(attributes.get(METADATA) == null) return null;
        return attributes.get(METADATA).getAsJsonObject();
    }
    public String getMetadata(String key){
        if(attributes.get(METADATA) == null) return null;
        return attributes.get(METADATA).getAsJsonObject().get(key).getAsString();
    }

    public void setMetadata(JsonObject jo){
        this.addAttribute(METADATA,jo);
    }
}

public class MetaDO extends DOWithMetadata {

    public MetaDO(String id){
        super(id,DoType.Metadata);
    }

    public static MetaDO fromDO(DigitalObject originalDO){
        MetaDO meta = new MetaDO(originalDO.id);
        if(originalDO.attributes == null || originalDO.attributes.get(METADATA) == null){
            return null;
        }
        meta.addAttribute(METADATA,originalDO.attributes.get(METADATA));
        return meta;
    }

    public static MetaDO fromDOWithMetadata(DOWithMetadata doWithMetadata){
        MetaDO meta = new MetaDO(doWithMetadata.id);
        meta.addAttribute(METADATA,doWithMetadata.getMetadata());
        return meta;
    }
}

public class DOWithOwner extends DigitalObject{

    public static final String DO_OWNER = "doOwner";

    public static DOWithOwner fromDO(DigitalObject digitalObject){
        if(digitalObject.attributes == null || digitalObject.attributes.get(DO_OWNER) == null){
            return null;
        }
        return (DOWithOwner)digitalObject;
    }

    public DOWithOwner(String id, DoType type) {
        super(id, type);
    }

    public void setDoOwner(String ownerID){
        addAttribute(DO_OWNER,ownerID);
    }

    public String getDoOwner(){
        if(attributes.get(DO_OWNER) == null) return null;
        return attributes.get(DO_OWNER).getAsString();
    }
}

public class BDRepositoryHandler extends RegistryHandlerBase {
    static Logger logger = Logger.getLogger(BDRepositoryHandler.class);
    BDRepoStorage store;        //数字对象的持久化存储，基于RocksDB
    DoIndexer indexer;          //本地创建数字对象索引，基于Lucene
    HashMap<String,RemoteReg> remoteRegs;       //记录该Repo接入的远程Registry，支持基于Registry的动态组网
    static final String REMOTE_REG = "REMOTEREG";
    static final String DO_OWNER = "doOwner";

    public BDRepositoryHandler(DoipServiceInfo info, BDRepoStorage store) throws RocksDBException {
        super(info);
        this.store = store;
        if(store.getRaw(REMOTE_REG.getBytes()) != null){
            remoteRegs = loadRegList();
        } else
            remoteRegs = new HashMap<>();
    }

    public BDRepositoryHandler(DoipServiceInfo info, BDRepoStorage store, DoIndexer indexer) throws RocksDBException {
        this(info,store);
        this.indexer = indexer;
    }

    @Override
    public DoipMessage handleHello(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Delete 的消息
    }

    @Override
    public DoipMessage handleListOps(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Delete 的消息
    }

    @Override
    public DoipMessage handleCreate(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Delete 的消息
    }

    @Override
    public DoipMessage handleUpdate(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Delete 的消息
    }

    @Override
    public DoipMessage handleDelete(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Delete 的消息
    }

    @Override
    public DoipMessage handleRetrieve(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Retrieve 的消息
    }

    @Override
    public DoipMessage handleSearch(DoipMessage request) {
        //处理DoipMessageHeader.parameter中operation=0.DOIP/Op.Search的消息
    }
}

public class BDRegistryHandler extends RegistryHandlerBase {
    /*
        其他部分详见具体代码，和BDRepository代码类似
    */
    
    //扩展操作Op.Join，允许其他Repository或Registry接入，提供分布式搜索功能
    //DoipMessageHeader.parameter中operation=Op.Join
    @Op(op = BasicOperations.Extension,name = "Op.Join")
    public DoipMessage handleJoin(DoipMessage request){
        if(request.header.parameters.attributes == null ||request.header.parameters.attributes.get("repoID")== null){
            logger.info("invalid join request: repoID not found");
            return replyStringWithStatus(request, "invalid join request: repoID not found.", DoipResponseCode.Invalid);
        }
        String repoID = request.header.parameters.attributes.get("repoID").getAsString();
        String repoUrl = request.header.parameters.attributes.get("repoUrl").getAsString();
        remoteRegs.put(repoID,repoUrl);
        saveRegFederation();
        return replyString(request,"success");
    }

    //扩展操作Op.Quit，允许其他Repository或Registry接入，提供分布式搜索功能
    //DoipMessageHeader.parameter中operation=Op.Quit
    @Op(op = BasicOperations.Extension,name = "Op.Quit")
    public DoipMessage handleQuit(DoipMessage request){
        if(request.header.parameters.attributes == null ||request.header.parameters.attributes.get("repoID")== null){
            logger.info("invalid join request: repoID not found");
            return replyStringWithStatus(request, "invalid join request: repoID not found.", DoipResponseCode.Invalid);
        }
        String repoID = request.header.parameters.attributes.get("repoID").getAsString();
        remoteRegs.remove(repoID);
        saveRegFederation();
        return replyString(request,"success");
    }
}

public interface RepositoryHandler {

    @Op(op = BasicOperations.Hello)
    DoipMessage handleHello(DoipMessage request);

    @Op(op = BasicOperations.ListOps)
    DoipMessage handleListOps(DoipMessage request);

    @Op(op = BasicOperations.Create)
    DoipMessage handleCreate(DoipMessage request);

    @Op(op = BasicOperations.Update)
    DoipMessage handleUpdate(DoipMessage request);

    @Op(op = BasicOperations.Delete)
    DoipMessage handleDelete(DoipMessage request);

    @Op(op = BasicOperations.Retrieve)
    DoipMessage handleRetrieve(DoipMessage request);

}

public class RequestHandlerImpl implements DoipRequestHandler {

    Map<String, Method> handlers;
    static Logger logger = Logger.getLogger(NettyServerHandler.class);
    protected RepositoryHandler doipHandler;

    public RequestHandlerImpl(RepositoryHandler doipHandler) {
        handlers = new HashMap<>();
        this.doipHandler = doipHandler;
        Class handlerClass = doipHandler.getClass();

        while(handlerClass != Object.class){
            putDoipHandlerMethod(handlerClass);
            Class[] interfaces = handlerClass.getInterfaces();
            for (Class clz : interfaces) {
                putDoipHandlerMethod(clz);
            }
            handlerClass = handlerClass.getSuperclass();
        }
    }

    private void putDoipHandlerMethod(Class handlerClass) {
        Method[] methods = handlerClass.getDeclaredMethods();
        for (Method m : methods) {
            Op a = m.getAnnotation(Op.class);
            if (a != null) {
                if (a.op() != BasicOperations.Extension) {
                    putHandler(a.op().getName(), m);
                } else{
                    putHandler(a.name(), m);
                }
            }
        }
    }

    private void putHandler(String name, Method m) {
        if (handlers.containsKey(name)) {
            return;
        }
        logger.debug("[Register operation] name: " + name);
        m.setAccessible(true);
        handlers.put(name, m);
    }

    @Override
    public DoipMessage onRequest(DoipMessage msg) {
        String str = msg.header.parameters.operation;
        logger.debug("[Call operation] name: " + str);
        if (str != null) {
            Method m;
            m = handlers.get(str);
            if (m == null) m = handlers.get(BasicOperations.Unknown.getName());
            if (m != null) {
                try {
                    return (DoipMessage) m.invoke(doipHandler, msg);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        return null;
    }
}

public class BDRegistryHandler extends RegistryHandlerBase {
    /*
        .....
    */
    @Op(op = BasicOperations.Extension,name = "Op.Join")
    public DoipMessage handleJoin(DoipMessage request){
            if(request.header.parameters.attributes == null ||request.header.parameters.attributes.get("repoID")== null){
                logger.info("invalid join request: repoID not found");
                return replyStringWithStatus(request, "invalid join request: repoID not found.", DoipResponseCode.Invalid);
            }
            String repoID = request.header.parameters.attributes.get("repoID").getAsString();
            String repoUrl = request.header.parameters.attributes.get("repoUrl").getAsString();
            inRegs.put(repoID,repoUrl);
            saveRegFederation();
            return replyString(request,"success");
        }
    /*
        .....
    */
}

public class BlueToothDoipListener implements DoipListener  {

    private static final String TAG = "BlueToothDoipListener";
    private static Context ctx;
    private BluetoothServerSocket mSSocket;
    private boolean isListenning = true;
    private PacketMessageCodec codec;
    DoipRequestHandler doipRequestHandler;
    DoipListenerInfo btListenerInfo;

    public BlueToothDoipListener(Context c, DoipListenerInfo listenerInfo) {
        codec = new PacketMessageCodecImpl();
        ctx = c;
        btListenerInfo = listenerInfo;
    }

    @Override
    public void start() {
        log("start listener");
        try {
            listen();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void stop() {
        try {
            mSSocket.close();
            isListenning = false;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void setRequestHandler(DoipRequestHandler doipRequestHandler) {
        this.doipRequestHandler = doipRequestHandler;
    }


    /**
     * 监听客户端发起的连接
     */
    @SuppressLint("HardwareIds")
    private void listen() throws IOException {
        BluetoothAdapter bluetoothAdapter = BluetoothAdapter.getDefaultAdapter();
        if(bluetoothAdapter == null) {
            //the device doesn't support bluetooth
            Log.e("BlueTooth","do not support Bluetooth");
            return;
        }
        if(!bluetoothAdapter.isEnabled()) {
            Log.e("Bluetooth","start blue tooth first");
            return;
        }

        mSSocket = bluetoothAdapter.listenUsingRfcommWithServiceRecord("DoipBTService", AndroidInitializer.DOIP_SERVICE_UUID);
        log("try to listen, local address: " + bluetoothAdapter.getAddress());
        log("try to listen, local name: " + bluetoothAdapter.getName());
        btListenerInfo.url = "bluetooth://" + bluetoothAdapter.getAddress();

        while(isListenning){
            waitForConnect();
            log("no input for 10s, connection released");
        }
    }

    private void waitForConnect() throws IOException {
        log("wait for connect");
        BluetoothSocket connectedSocket = mSSocket.accept();
        log("device connected, name: " +connectedSocket.getRemoteDevice().getName());
        log("device connected, address: " +connectedSocket.getRemoteDevice().getAddress());
        new BtSocketThread(connectedSocket).start();
    }

    private void handleAndResponseDoipMessage(DoipMessage msg, OutputStream output){
        if (msg.header.parameters == null || msg.header.parameters.operation == null || msg.header.parameters.response != null) {
            replyStringWithStatus(output,msg,"invalid request", DoipResponseCode.Invalid);
            return;
        };
        if(msg.header.isCertified()){
            try {
                if(!GlobalCertifications.verifyDoipMessage(msg)){
                    log("verification failed");
                    return;
                }
            }catch (Exception e){
                e.printStackTrace();
                return;
            }
        }
        if (msg.credential != null) log("[Caller] client ID: " + msg.credential.getSigner());
        else log("[Caller] client ID: Anonymous");

        DoipMessage response = doipRequestHandler.onRequest(msg);

        if(response != null)
            sendResponse(output, response);
        else
            defaultHandler(output,msg);
    }

    private void defaultHandler(OutputStream ctx, DoipMessage request) {
        replyStringWithStatus(ctx,request,"Unsupported Operation!", DoipResponseCode.Declined);
    }

    private void replyStringWithStatus(OutputStream output, DoipMessage request, String str, DoipResponseCode resp) {
        DoipMessage response;
        response = new DoipMessageFactory.DoipMessageBuilder()
                .createResponse(resp,request)
                .setBody(str.getBytes())
                .create();
        sendResponse(output,response);
    }

    //基于编解码器将DoipMessage编码并通过Bluetooth协议的OutputStream发送
    private void sendResponse(OutputStream output, DoipMessage response){
        try {
            if(response.header.isCertified()){
                try {
                    GlobalCertifications.signDoipMessage(response);
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
            ArrayList<MessageEnvelope> envs = codec.MessageToEnvelopes(response);
            for(MessageEnvelope env:envs) {
                byte[] envBytes = codec.EnvelopeToBytes(env);
                log("sending bytes length: " + envBytes.length);
                output.write(codec.EnvelopeToBytes(env));
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    //基于编解码器从Bluetooth协议的InputStream中解码DoipMessage
    private DoipMessage getDoipMessage(InputStream in){
        DoipMessage message = null;
        try {
            byte[] envHeader = getBytesFromInputStream(in,24);
            int requestId = ((envHeader[8+3] & 0xFF)
                    | ((envHeader[8+2] & 0xFF)<<8)
                    | ((envHeader[8+1] & 0xFF)<<16)
                    | ((envHeader[8] & 0xFF)<<24));
            log("requestId: " + requestId);
            int contentLength = (int) ((envHeader[20+3] & 0xFF)
                    | ((envHeader[20+2] & 0xFF)<<8)
                    | ((envHeader[20+1] & 0xFF)<<16)
                    | ((envHeader[20] & 0xFF)<<24));
            log("content Length: " + contentLength);

            byte[] envContent = getBytesFromInputStream(in, contentLength);
            return codec.BytesToMessage(envContent,requestId,null);

        } catch (Exception e) {
            e.printStackTrace();
        }

        return message;
    }

    private byte[] getBytesFromInputStream(InputStream in, int len) throws IOException {
        log("reading inputStream, available: " + in.available());
        int count = 0;
        while(in.available() < len && count <100){
            count ++;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        byte[] bytes = new byte[len];
        in.read(bytes);
        return bytes;
    }

    private static void log(String s){
        Log.d(TAG,s);
    }


    class BtSocketThread extends Thread{
        BluetoothSocket socket;
        BtSocketThread(BluetoothSocket s){
            socket = s;
        }

        @Override
        public void run(){
            try {
                int missingInputCount = 0;
                InputStream input = socket.getInputStream();
                OutputStream output = socket.getOutputStream();
                while(socket.isConnected()){
                    if(input.available() == 0){
                        missingInputCount ++;
//                        if(missingInputCount > 10){
//                            socket.close();
//                        }
                        try {
                            Thread.sleep(500);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        continue;
                    }
                    DoipMessage req = getDoipMessage(input);
                    handleAndResponseDoipMessage(req,output);
                }
                log("device disconnected, name: " + socket.getRemoteDevice().getName());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

public class BlueToothDoipClient implements DoipClientChannel {

    private Context ctx;
    private String address;
    private final String TAG = "BluetoothDoipClient";

    private BluetoothDevice device;
    private BluetoothAdapter bluetoothAdapter;
    private BluetoothSocket targetSocket;
    private PacketMessageCodec codec = new PacketMessageCodecImpl();
    Random random = new Random();
    boolean isConnected = false;

    private HashMap<Integer,DoipMessageCallback> waitingCallback = new HashMap<>();

    private boolean connecting = false;

    public BlueToothDoipClient(Context c){
        ctx = c;
    }

    @Override
    public void close(){
        try {
            targetSocket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void connect(String url) throws URISyntaxException{
        isConnected = false;
        URI uri = new URI(url);
        this.address = uri.getAuthority();
        try {
            connectBTServer();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public boolean isConnected() {
        return isConnected;
    }

    @Override
    public void sendMessage(DoipMessage message, DoipMessageCallback cb){

        log("try to send message");
        if(device == null){
            Log.e(TAG,"no connection to target repository device");
        }
        try {
            while(!isConnected){
                log("not connected, wait for connect.");
                Thread.sleep(500);
            }

            if(message.requestID == 0) message.requestID = random.nextInt();
            log("requestID in message: " + message.requestID);

            OutputStream output = targetSocket.getOutputStream();
            ArrayList<MessageEnvelope> envelopes = codec.MessageToEnvelopes(message);

            waitingCallback.put(message.requestID,cb);

            log("connected: " + targetSocket.isConnected());
            for (MessageEnvelope env : envelopes) {
                log("requestID in envelope: " + env.requestId);
                byte[] envBytes = codec.EnvelopeToBytes(env);
                log("sending bytes length: " + envBytes.length);
                output.write(codec.EnvelopeToBytes(env));
            }
            InputStream input = targetSocket.getInputStream();
            new ResponseHandlerThread(input).start();

        }catch (Exception e){
            e.printStackTrace();
            targetSocket = null;
        }
    };


    private void connectBTServer() throws IOException {

        bluetoothAdapter = BluetoothAdapter.getDefaultAdapter();
        if (bluetoothAdapter == null) {
            // 说明此设备不支持蓝牙操作
            log("do not support Bluetooth");
            return;
        }
        if (!bluetoothAdapter.isEnabled()) {
            AlertDialog dialog = new AlertDialog.Builder(ctx).setTitle("提示").setMessage("打开蓝牙再尝试!").create();
            dialog.show();
            return;
        }
        for(BluetoothDevice dev:bluetoothAdapter.getBondedDevices()){
            log("bounded bt address: " + dev.getAddress());
            log("bounded bt name: " + dev.getName());
            log("bounded bt state: " + dev.getType());
        };

        device = bluetoothAdapter.getRemoteDevice(address);
        if(device == null){
            log("cannot find target device. ");
            return;
        }
        log("remote bt address: " + device.getAddress());
        log("remote bt name: " + device.getName());
        log("remote bt state: " + device.getType());

        targetSocket = device.createRfcommSocketToServiceRecord(AndroidInitializer.DOIP_SERVICE_UUID);

        new ConnectThread(targetSocket).start();
    }

    void log(String s){
        Log.d(TAG,s);
    }

    class ConnectThread extends Thread{
        private BluetoothSocket mSocket;

        ConnectThread(BluetoothSocket s) throws IOException {
            mSocket = s;
        }

        @Override
        public synchronized void run(){
            // 关闭发现设备
            bluetoothAdapter.cancelDiscovery();
            if(mSocket.isConnected()) return;
            try{
                log("try to connect device");
                connecting = true;
                mSocket.connect();
                log("connected: " + mSocket.isConnected());
                log("remote device name: " + mSocket.getRemoteDevice().getName());
                log("remote device address: " + mSocket.getRemoteDevice().getAddress());
                isConnected = true;
            }catch(IOException connectException){
                connectException.printStackTrace();
            }finally {
                connecting = false;
            }
        }
    }

    class ResponseHandlerThread extends Thread{
        InputStream input;
        ResponseHandlerThread(InputStream in){
            input = in;
        }

        @Override
        public synchronized void run(){
            log("try to get response:");
            DoipMessage resp = getDoipMessageFromInputStream(input);
            if(resp == null){
                Log.e(TAG,"Unable to get response message");
                return;
            }
            if(waitingCallback.get(resp.requestID) == null){
                Log.e(TAG,"Unable to get callback");
                return;
            }
            waitingCallback.get(resp.requestID).onResult(resp);
            waitingCallback.remove(resp.requestID);
        }

    }
    private DoipMessage getDoipMessageFromInputStream(InputStream in){
        DoipMessage message = null;
        try {
            byte[] envHeader = getBytesFromInputStream(in,24);
            int requestId = (int) ((envHeader[8+3] & 0xFF)
                    | ((envHeader[8+2] & 0xFF)<<8)
                    | ((envHeader[8+1] & 0xFF)<<16)
                    | ((envHeader[8] & 0xFF)<<24));
            log("requestID: " + requestId);
            int contentLength = (int) ((envHeader[23] & 0xFF)
                    | ((envHeader[22] & 0xFF)<<8)
                    | ((envHeader[21] & 0xFF)<<16)
                    | ((envHeader[20] & 0xFF)<<24));
            log("content Length: " + contentLength);

            byte[] envContent = getBytesFromInputStream(in, contentLength);
            return codec.BytesToMessage(envContent,requestId,null);

        } catch (Exception e) {
            e.printStackTrace();
        }

        return message;
    }

    private byte[] getBytesFromInputStream(InputStream in, int len) throws IOException {
        int count = 0;
        while(in.available() < len && count <100){
            count ++;
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        byte[] bytes = new byte[len];
        in.read(bytes);
        return bytes;
    }
}

{
    "id": "创建的时候为空",
    "type": "0.TYPE/DO",
    "attributes": {
        "desc": "属性信息",
        "metadata": {
            "desc": "元数据相关信息"
        }
    },
    "elements": [
        {
            "id": "DO内部唯一的ID",
            "length": "元素长度",
            "type": "string",
            "dataString": "元素data字节数组的字符串形式",
            "attributes": {
                "desc": "元素属性"
            },
            "data": []
        }
    ]
}

 DoipMessage msg =
                new DoipMessageFactory.DoipMessageBuilder()
                        .createRequest(doipInfo.id, BasicOperations.Create.getName())
                        .setBody(digitalObject)
                        .create();
//设置用户标识
msg.credential = new MessageCredential(userID);

 DoipMessage msg =
                new DoipMessageFactory.DoipMessageBuilder()
                        .createRequest(doID,BasicOperations.Retrieve.getName())
                        .create();

msg.header.parameters.addAttribute("includeElementData", true);

msg.header.parameters.addAttribute("element", elementID);