dubbo接口服务调用公共类

import json, time
import telnetlib
from datetime import datetime


class Dubbo(telnetlib.Telnet):
    prompt = 'dubbo>'
    coding = 'utf-8'

    def __init__(self, host=None, port=0):
        super().__init__(host, port)
        # 触发dubbo提示符
        self.write(b'
')

    def command(self, flag, str_=""):
        data = self.read_until(flag.encode())
        self.write(str_.encode() + b"
")
        return data

    def invoke(self, service_name, method_name, arg):
        self.command(Dubbo.prompt, "invoke {0}.{1}({2})".format(service_name, method_name, json.dumps(arg)))
        data = self.command(Dubbo.prompt, "")
        data = data.decode(Dubbo.coding, errors='ignore').split('
')[0].strip()

        # 终止Telnet连接
        self.close()
        return data

    def do(self, arg):
        self.command(Dubbo.prompt, arg)
        data = self.command(Dubbo.prompt, arg)
        data = data.decode(Dubbo.coding, errors='ignore').split('
')[0].strip()

        # 终止Telnet连接
        self.close()
        return data

参考资料百度，某人博客摘抄，忘记了不好意思