<|start_filename|>demo/api/src/main/java/com/colobu/rpcx/service/ITestService.java<|end_filename|>
package com.colobu.rpcx.service;

import com.colobu.rpcx.rpc.annotation.Consumer;

/**
 * @author <EMAIL>
 */
@Consumer(impl = "com.colobu.rpcx.service.TestService")
public interface ITestService {

    String hi(String str);


    byte[] golangHi(byte[] data);


    int sum(int a, int b);


    int sum2(int a, int b);

    /**
     * 异步调用
     * @return
     */
    String async();

}


<|start_filename|>client/src/test/java/com/colobu/rpcx/client/NettyClientTest.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.FilterWrapper;
import com.colobu.rpcx.protocol.*;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.impl.RpcConsumerInvoker;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.junit.Test;

import java.util.stream.IntStream;

/**
 * Created by zhangzhiyong on 2018/7/3.
 */
public class NettyClientTest {


    /**
     * 调用golang 并且使用filter增强
     */
    @Test
    public void testCallGolang() {
        ZkServiceDiscovery serviceDiscovery = new ZkServiceDiscovery("/youpin/services/", "Arith");
        NettyClient client = new NettyClient(serviceDiscovery);
        Invoker invoker = new RpcConsumerInvoker(client);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.CONSUMER, Sets.newHashSet());

        RpcInvocation invocation = new RpcInvocation();
        invocation.setMethodName("Echo");
        invocation.setClassName("Arith");
        invocation.setLanguageCode(LanguageCode.GO);
        invocation.setPayload("zzy".getBytes());
        invocation.setUrl(new URL("rpcx", "", 0));
        invocation.setSerializeType(SerializeType.SerializeNone);

        Result result = wrapperInvoker.invoke(invocation);

        if (!result.hasException()) {
            System.out.println(new String((byte[]) (result.getValue())));
        } else {
            result.getException().printStackTrace();
        }
    }


    /**
     * 直连
     *
     * @throws Exception
     */
    @Test
    public void testSendMsg() throws Exception {
        Message req = new Message("Arith", "Echo");
        req.setVersion((byte) 0);
        req.setMessageType(MessageType.Request);
        req.setHeartbeat(false);
        req.setOneway(false);
        req.setCompressType(CompressType.None);
        req.setSerializeType(SerializeType.SerializeNone);
        req.setSeq(123);
        req.metadata.put("test", "1234");
        req.payload = "world".getBytes("UTF-8");

        NettyClient client = new NettyClient(null);
        Message res = client.call("192.168.31.82:8997", req, 1000);
        System.out.println(new String(res.payload));
    }

    //经过服务发现
    @Test
    public void testGolangServiceFinder() throws Exception {
        Message req = new Message("Arith", "Echo");
        req.setVersion((byte) 0);
        req.setMessageType(MessageType.Request);
        req.setHeartbeat(false);
        req.setOneway(false);
        req.setCompressType(CompressType.None);
        req.setSerializeType(SerializeType.SerializeNone);
        req.setSeq(123);
        req.metadata.put("test", "1234");
        req.payload = "world".getBytes("UTF-8");

        ZkServiceDiscovery serviceDiscovery = new ZkServiceDiscovery("/youpin/services/", "Arith");//发现golang的服务
        NettyClient client = new NettyClient(serviceDiscovery);
        Message res = client.call(req, 2000, Constants.SYNC_KEY);
        System.out.println(new String(res.payload));
    }


    //测试循环发送消息
    @Test
    public void testLoopSendMsg() throws Exception {
        Message req = new Message("Arith", "Echo");
        req.setVersion((byte) 0);
        req.setMessageType(MessageType.Request);
        req.setHeartbeat(false);
        req.setOneway(false);
        req.setCompressType(CompressType.None);
        req.setSerializeType(SerializeType.SerializeNone);
        req.setSeq(123);
        req.metadata.put("test", "1234");
        req.payload = "world".getBytes("UTF-8");

        NettyClient client = new NettyClient(null);
        IntStream.range(0, 100).forEach(it -> {
            Message res = null;
            try {
                res = client.call("10.231.72.75:8976", req, 1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            System.out.println(new String(res.payload));
        });
    }
}


<|start_filename|>demo/democlient/src/test/java/com/colobu/rpcx/test/ClientTest.java<|end_filename|>
package com.colobu.rpcx.test;


import com.colobu.rpcx.bootstrap.Bootstrap;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Bootstrap.class)
public class ClientTest {


    @Test
    public void testStartSpring() {

    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/annotation/Provider.java<|end_filename|>
package com.colobu.rpcx.rpc.annotation;

import java.lang.annotation.*;


/**
 * Created by <EMAIL>.
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Documented
public @interface Provider {

    String group() default "";

    String name() default "";

    String version() default "";

    String[] excludeFilters() default "";

    /**
     * 服务启动token验证,客户端需要提供相同的token
     *
     * @return
     */
    String token() default "";

    /**
     * 服务器超时调用后会打印警告,不影响正常调用
     * 超时filter
     *
     * @return
     */
    long timeout() default -1;

    /**
     * 缓存filter
     *
     * @return
     */
    boolean cache() default false;

    /**
     * 是否开启监控
     * 监控filter
     *
     * @return
     */
    boolean monitor() default false;

    /**
     * 一分钟内tps 能达到的上限  -1 是没有上限
     * tps filter
     *
     * @return
     */
    int tps() default -1;

    /**
     * 权重
     *
     * @return
     */
    String weight() default "";

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/Invoker.java<|end_filename|>
package com.colobu.rpcx.rpc;

import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.esotericsoftware.reflectasm.MethodAccess;

import java.lang.reflect.Method;


/**
 * @author <EMAIL>
 */
public interface Invoker<T> extends Node {

    Class<T> getInterface();

    Result invoke(RpcInvocation invocation) throws RpcException;

    void setMethod(Method method);

    default void setMethodAccess(MethodAccess methodAccess){

    }

    Method getMethod();

    void setInterface(Class clazz);

    /**
     * 服务发现
     * @return
     */
    default IServiceDiscovery serviceDiscovery() {
        return null;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/EchoFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.rpc.impl.RpcResult;

/**
 * @author <EMAIL>
 */
@RpcFilter(order = -2001, group = {Constants.PROVIDER})
public class EchoFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation inv) throws RpcException {
        if (inv.getMethodName().equals(Constants.$ECHO) && inv.getArguments() != null && inv.getArguments().length == 1) {
            return new RpcResult(inv.getArguments()[0]);
        }
        return invoker.invoke(inv);
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/monitor/Monitor.java<|end_filename|>
package com.colobu.rpcx.monitor;

import com.colobu.rpcx.rpc.Node;

/**
 * Created by <EMAIL>.
 */
public interface Monitor extends Node, MonitorService {

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/FailFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;

import com.colobu.rpcx.common.retry.RetryNTimes;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.fail.FailType;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.selector.SelectMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author <EMAIL>
 * <p>
 * 负责管理失败的重试
 */
@RpcFilter(group = {Constants.CONSUMER}, order = -1000)
public class FailFilter implements Filter {


    private static final Logger logger = LoggerFactory.getLogger(FailFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        FailType failType = invocation.getFailType();
        RetryNTimes retry = new RetryNTimes(invocation.getRetryNum());
        Result result = retry.retry((n) -> {
            Result res = invoker.invoke(invocation);
            if (res.hasException()) {
                logger.warn("{} invoke error:{}", invocation.getUrl().getPath(), res.getException().getMessage());
                //失败了立刻返回
                if (failType.equals(FailType.FailFast)) {
                    logger.info("------>failFast");
                    res.getAttachments().put("needRetry", "false");
                    return res;
                }
                //换个服务节点重试
                if (failType.equals(FailType.FailOver)) {
                    logger.info("------>failOver");
                    res.getAttachments().put("needRetry", "true");
                    return res;
                }
                //当前节点重试
                if (failType.equals(FailType.FailTry)) {
                    logger.info("------>failTry");
                    res.getAttachments().put("needRetry", "true");
                    //这样走到下游的SelectFilter就不会再选出来一个新的Service了
                    invocation.setSelectMode(SelectMode.SelectByUser);
                    return res;
                }
            }
            res.getAttachments().put("needRetry", "false");
            return res;
        });

        return result;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/Pair.java<|end_filename|>
package com.colobu.rpcx.common;

import java.util.Objects;

/**
 * @author <EMAIL>
 */
public class Pair<T1, T2> {
    private T1 object1;
    private T2 object2;


    public Pair(T1 object1, T2 object2) {
        this.object1 = object1;
        this.object2 = object2;
    }


    public static <T1, T2> Pair<T1, T2> of(T1 object, T2 object2) {
        return new Pair<>(object, object2);
    }


    public T1 getObject1() {
        return object1;
    }


    public void setObject1(T1 object1) {
        this.object1 = object1;
    }


    public T2 getObject2() {
        return object2;
    }


    public void setObject2(T2 object2) {
        this.object2 = object2;
    }


    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (!(o instanceof Pair)) {
            return false;
        }
        Pair<?, ?> pair = (Pair<?, ?>) o;
        return Objects.equals(getObject1(), pair.getObject1()) &&
                Objects.equals(getObject2(), pair.getObject2());
    }

    @Override
    public int hashCode() {

        return Objects.hash(getObject1(), getObject2());
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/cache/CacheFactory.java<|end_filename|>
package com.colobu.rpcx.cache;


import com.colobu.rpcx.rpc.URL;

/**
 * @author <EMAIL>
 */
public interface CacheFactory {

    Cache getCache(URL url);

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/ConsumerFinder.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.rpc.annotation.Consumer;
import com.colobu.rpcx.rpc.annotation.Provider;
import org.reflections.Reflections;

import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author <EMAIL>
 */
public class ConsumerFinder {

    public Set<String> find(String consumerPackage) {
        Reflections reflections = new Reflections(consumerPackage);
        //不包括实现类
        Set<Class<?>> classesList = reflections.getTypesAnnotatedWith(Consumer.class,true);
        return  classesList.stream().map(it->{
            Consumer consumer = it.getAnnotation(Consumer.class);
            return consumer.impl();
        }).collect(Collectors.toSet());
    }

}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/NettyClient.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.common.*;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.config.NettyClientConfig;
import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.exception.RemotingSendRequestException;
import com.colobu.rpcx.exception.RemotingTimeoutException;
import com.colobu.rpcx.exception.RemotingTooMuchRequestException;
import com.colobu.rpcx.netty.*;
import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.rpc.RpcContext;
import com.colobu.rpcx.rpc.RpcException;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author <EMAIL>
 */
public class NettyClient extends NettyRemotingAbstract implements IClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private DefaultEventExecutorGroup defaultEventExecutorGroup;

    private NettyClientConfig nettyClientConfig = new NettyClientConfig();

    private final EventLoopGroup eventLoopGroupWorker;

    private final long CallTimeOut = TimeUnit.SECONDS.toMillis(3);

    protected final NettyEventExecuter nettyEventExecuter = new NettyEventExecuter();

    private Bootstrap bootstrap;

    private final IServiceDiscovery serviceDiscovery;

    protected final Semaphore semaphoreAsync;

    protected final Semaphore semaphoreOneway;


    public NettyClient(IServiceDiscovery serviceDiscovery) {
        this.semaphoreOneway = new Semaphore(1000, true);
        this.semaphoreAsync = new Semaphore(1000, true);
        this.serviceDiscovery = serviceDiscovery;
        this.eventLoopGroupWorker = new NioEventLoopGroup(1, new NamedThreadFactory("NettyClientSelector_"));
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(nettyClientConfig.getClientWorkerThreads(), new NamedThreadFactory("NettyClientWorkerThread_"));

        this.bootstrap = createBootstrap();

        startScanResponseTableSchedule();
        runEventListener();
    }

    private void runEventListener() {
        if (this.channelEventListener != null) {
            new Thread(this.nettyEventExecuter).start();
        }
    }

    private Bootstrap createBootstrap() {
        return new Bootstrap().group(this.eventLoopGroupWorker).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
                .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(
                                defaultEventExecutorGroup,
                                new NettyEncoder(),
                                new NettyDecoder(),
                                //*空闲状态的handler
                                new IdleStateHandler(0, 0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                                //管理连接的
                                new NettyConnetManageHandler(NettyClient.this),
                                //处理具体业务逻辑的handler
                                new NettyClientHandler(NettyClient.this));
                    }
                });
    }

    public void startScanResponseTableSchedule() {
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("ClientHouseKeepingService"));
        executor.scheduleAtFixedRate(() -> {
            try {
                NettyClient.this.scanResponseTable();//* 查询response表
            } catch (Exception e) {
                logger.error("scanResponseTable exception", e);
            }
        }, 3000, 3000, TimeUnit.MILLISECONDS);
    }


    @Override
    public Message call(String addr, Message req, long timeOut) throws Exception {
        final RemotingCommand request = RemotingCommand.createRequestCommand(req);
        long timeoutMillis = timeOut;
        //获取或者创建channel
        final Channel channel = this.getAndCreateChannel(addr);
        //同步执行
        RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis);
        return response.getMessage();
    }


    @Override
    public Message call(Message req, long timeoutMillis, String sendType) throws Exception {
        final RemotingCommand request = RemotingCommand.createRequestCommand(req);

        Channel channel = null;
        String serviceAddr = RpcContext.getContext().getServiceAddr();
        logger.info("---------||||---->serviceAddr:{}", serviceAddr);
        if (StringUtils.isEmpty(serviceAddr)) {
            throw new RpcException("service addr is null  call method:" + req.getServiceMethod());
        } else {
            channel = this.getAndCreateChannel(serviceAddr);
        }
        setHostAndPort(req, channel);
        Message res = null;
        if (StringUtils.isEmpty(sendType)) {
            sendType = Constants.SYNC_KEY;
        }
        if (sendType.equals(Constants.SYNC_KEY)) {
            RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis);
            res = response.getMessage();
        } else if (sendType.equals(Constants.ASYNC_KEY)) {
            ResponseFuture future = this.invokeAsyncImpl(channel, request, timeoutMillis, null);
            res = new Message();
            RpcContext.getContext().setFuture(future);
        } else if (sendType.equals(Constants.ONE_WAY_KEY)) {
            this.invokeOnewayImpl(channel, request, timeoutMillis);
            res = new Message();
        }
        return res;
    }

    private void setHostAndPort(Message req, Channel channel) {
//        String host = "";
//        int port = 0;
//        if (channel.localAddress() instanceof InetSocketAddress) {
//            host = ((InetSocketAddress) channel.localAddress()).getAddress().toString();
//            port = ((InetSocketAddress) channel.localAddress()).getPort();
//        }

//        req.metadata.put("_host", host);
//        req.metadata.put("_port", String.valueOf(port));
    }


    private Channel getAndCreateChannel(final String addr) throws InterruptedException {
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }
        return this.createChannel(addr);
    }


    private Channel createChannel(final String addr) throws InterruptedException {
        ChannelWrapper cw = this.channelTables.get(addr);
        if (cw != null && cw.isOK()) {
            return cw.getChannel();
        }

        if (this.lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
            try {
                boolean createNewConnection = false;
                cw = this.channelTables.get(addr);
                if (cw != null) {
                    if (cw.isOK()) {
                        return cw.getChannel();
                    } else if (!cw.getChannelFuture().isDone()) {
                        createNewConnection = false;
                    } else {
                        this.channelTables.remove(addr);
                        createNewConnection = true;
                    }
                } else {
                    createNewConnection = true;
                }
                //需要创建新的连接
                if (createNewConnection) {
                    ChannelFuture channelFuture = this.bootstrap.connect(string2SocketAddress(addr));
                    logger.info("createChannel: begin to connect remote host[{}] asynchronously", addr);
                    cw = new ChannelWrapper(channelFuture);
                    this.channelTables.put(addr, cw);
                }
            } catch (Exception e) {
                logger.error("createChannel: create channel exception", e);
            } finally {
                this.lockChannelTables.unlock();
            }
        } else {
            logger.warn("createChannel: try to lock channel table, but timeout, {}ms", LockTimeoutMillis);
        }

        if (cw != null) {
            ChannelFuture channelFuture = cw.getChannelFuture();
            if (channelFuture.awaitUninterruptibly(this.nettyClientConfig.getConnectTimeoutMillis())) {
                if (cw.isOK()) {
                    logger.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                    return cw.getChannel();
                } else {
                    logger.warn("createChannel: connect remote host[" + addr + "] failed, " + channelFuture.toString(), channelFuture.cause());
                }
            } else {
                logger.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.nettyClientConfig.getConnectTimeoutMillis(),
                        channelFuture.toString());
            }
        }
        return null;
    }


    public static SocketAddress string2SocketAddress(final String addr) {
        String[] s = addr.split(":");
        InetSocketAddress isa = new InetSocketAddress(s[0], Integer.parseInt(s[1]));
        return isa;
    }

    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        request.markOnewayRPC();
        boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
            try {
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) {
                        once.release();//* 释放一次限流次数
                        if (!f.isSuccess()) {
                            logger.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
                        }
                    }
                });
            } catch (Exception e) {
                once.release();
                logger.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {
            if (timeoutMillis <= 0) {
                throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
            } else {
                String info = String.format(
                        "invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                        timeoutMillis,
                        this.semaphoreAsync.getQueueLength(),
                        this.semaphoreAsync.availablePermits()
                );
                logger.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }
    }


    /**
     * 异步调用
     *
     * @param channel
     * @param request
     * @param timeoutMillis
     * @param invokeCallback
     * @return
     * @throws InterruptedException
     * @throws RemotingTooMuchRequestException
     * @throws RemotingSendRequestException
     */
    public ResponseFuture invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
                                          final InvokeCallback invokeCallback)
            throws InterruptedException, RemotingTooMuchRequestException, RemotingSendRequestException {
        final int opaque = request.getOpaque();
        //达到限流作用
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            //用来保证只释放1次的
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);

            final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, invokeCallback, once);
            this.responseTable.put(opaque, responseFuture);
            try {
                channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else {
                        responseFuture.setSendRequestOK(false);
                    }
                    //发送失败会解除阻塞
                    responseFuture.putResponse(null);
                    responseTable.remove(opaque);
                    try {
                        responseFuture.executeInvokeCallback();
                    } catch (Throwable e) {
                        logger.warn("excute callback in writeAndFlush addListener, and callback throw", e);
                    } finally {
                        responseFuture.release();
                    }

                    logger.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
                });
            } catch (Exception e) {
                responseFuture.release();
                logger.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
            return responseFuture;
        } else {
            String info =
                    String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                            timeoutMillis,
                            this.semaphoreAsync.getQueueLength(),
                            this.semaphoreAsync.availablePermits()
                    );
            logger.warn(info);
            throw new RemotingTooMuchRequestException(info);
        }
    }


    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        final int opaque = request.getOpaque();

        try {
            final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, null, null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            //通过网络发送信息
            channel.writeAndFlush(request).addListener((ChannelFutureListener) f -> {
                if (f.isSuccess()) {
                    responseFuture.setSendRequestOK(true);
                    return;
                } else {
                    responseFuture.setSendRequestOK(false);
                }
                //发送失败responseTable中直接删除
                responseTable.remove(opaque);
                responseFuture.setCause(f.cause());
                responseFuture.putResponse(null);
                logger.warn("send a request command to channel <" + addr + "> failed.");
            });
            //这里会直接阻塞住(CountDownLatch)
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) {
                if (responseFuture.isSendRequestOK()) {
                    throw new RemotingTimeoutException(addr.toString(), timeoutMillis,
                            responseFuture.getCause());
                } else {
                    throw new RemotingSendRequestException(addr.toString(), responseFuture.getCause());
                }
            }
            return responseCommand;
        } finally {
            this.responseTable.remove(opaque);
        }
    }


    @Override
    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecuter.putNettyEvent(event);
    }

    @Override
    public IServiceDiscovery getServiceDiscovery() {
        return serviceDiscovery;
    }

    @Override
    public void close() {
        logger.info("netty client close");
        if (null != this.eventLoopGroupWorker) {
            this.eventLoopGroupWorker.shutdownGracefully();
        }

        if (this.nettyEventExecuter != null) {
            this.nettyEventExecuter.shutdown();
        }

        if (this.defaultEventExecutorGroup != null) {
            this.defaultEventExecutorGroup.shutdownGracefully();
        }

        if (null != this.serviceDiscovery) {
            this.serviceDiscovery.close();
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/TokenFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcContext;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;

import java.util.Map;

/**
 * Created by <EMAIL>.
 */
@RpcFilter(order = -999, group = {Constants.PROVIDER})
public class TokenFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation inv)
            throws RpcException {
        String token = invoker.getUrl().getParameter(Constants.TOKEN_KEY,"");
        if (!StringUtils.isEmpty(token)) {
            Class<?> serviceType = invoker.getInterface();
            Map<String, String> attachments = inv.getAttachments();
            String remoteToken = attachments == null ? null : attachments.get(Constants.TOKEN_KEY);
            if (!token.equals(remoteToken)) {
                throw new RpcException("Invalid token! Forbid invoke remote service " + serviceType + " method " + inv.getMethodName() + "() from consumer " + RpcContext.getContext().getRemoteHost() + " to provider " + RpcContext.getContext().getLocalHost());
            }
        }
        return invoker.invoke(inv);
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyServerHandler.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.protocol.RemotingCommand;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * @author <EMAIL>
 */
public class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {

    private final NettyRemotingAbstract nettyServer;

    public NettyServerHandler(NettyRemotingAbstract nettyClient) {
        this.nettyServer = nettyClient;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) {
        nettyServer.processMessageReceived(ctx, msg);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/AccessLogFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.common.NamedThreadFactory;
import com.colobu.rpcx.concurrent.ConcurrentHashSet;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.*;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileWriter;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author <EMAIL>
 */
@RpcFilter(group = {Constants.PROVIDER},order = -999)
public class AccessLogFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(AccessLogFilter.class);

    private static final String FILE_DATE_FORMAT = "yyyyMMdd";

    private static final String MESSAGE_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";

    private static final int LOG_MAX_BUFFER = 5000;

    private static final long LOG_OUTPUT_INTERVAL = 5000;

    private final static ConcurrentMap<String, Set<String>> LOG_QUEUE = new ConcurrentHashMap<>();

    private final static ScheduledExecutorService LOG_SCHEDULED = Executors.newScheduledThreadPool(2, new NamedThreadFactory("Rcpx-Access-Log", true));

    private volatile ScheduledFuture<?> logFuture = null;

    /**
     * 非阻塞
     */
    private class LogTask implements Runnable {
        @Override
        public void run() {
            try {
                if (LOG_QUEUE != null && LOG_QUEUE.size() > 0) {
                    for (Map.Entry<String, Set<String>> entry : LOG_QUEUE.entrySet()) {
                        try {
                            String accesslog = entry.getKey();
                            Set<String> logSet = entry.getValue();
                            File file = new File(accesslog);
                            File dir = file.getParentFile();
                            if (null != dir && !dir.exists()) {
                                dir.mkdirs();
                            }
                            if (file.exists()) {
                                String now = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date());
                                String last = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date(file.lastModified()));
                                if (!now.equals(last)) {
                                    File archive = new File(file.getAbsolutePath() + "." + last);
                                    file.renameTo(archive);
                                }
                            }
                            FileWriter writer = new FileWriter(file, true);
                            try {
                                for (Iterator<String> iterator = logSet.iterator();
                                     iterator.hasNext();
                                     iterator.remove()) {
                                    writer.write(iterator.next());
                                    writer.write("\r\n");
                                }
                                writer.flush();
                            } finally {
                                writer.close();
                            }
                        } catch (Exception e) {
                            logger.error(e.getMessage(), e);
                        }
                    }
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    private void init() {
        if (logFuture == null) {
            synchronized (LOG_SCHEDULED) {
                if (logFuture == null) {
                    logFuture = LOG_SCHEDULED.scheduleWithFixedDelay(new LogTask(), LOG_OUTPUT_INTERVAL, LOG_OUTPUT_INTERVAL, TimeUnit.MILLISECONDS);
                }
            }
        }
    }

    private void log(String accesslog, String message) {
        init();
        Set<String> logSet = LOG_QUEUE.get(accesslog);
        if (logSet == null) {
            LOG_QUEUE.putIfAbsent(accesslog, new ConcurrentHashSet<>());
            logSet = LOG_QUEUE.get(accesslog);
        }
        if (logSet.size() < LOG_MAX_BUFFER) {
            logSet.add(message);
        }
    }

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation inv) throws RpcException {
        if (inv.getUrl().getParameter(Constants.ACCESSLOG_KEY,Constants.FALSE).equals(Constants.FALSE)) {
            return invoker.invoke(inv);
        }

        try {
            RpcContext context = RpcContext.getContext();
            String serviceName = invoker.getInterface().getName();
            String version = invoker.getUrl().getParameter(Constants.VERSION_KEY);
            String group = invoker.getUrl().getParameter(Constants.GROUP_KEY);
            StringBuilder sn = new StringBuilder();
            sn.append("[").append(new SimpleDateFormat(MESSAGE_DATE_FORMAT).format(new Date())).append("] ").append(context.getRemoteHost()).append(":").append(context.getRemotePort())
                    .append(" -> ").append(context.getLocalHost()).append(":").append(context.getLocalPort())
                    .append(" - ");
            if (null != group && group.length() > 0) {
                sn.append(group).append("/");
            }
            sn.append(serviceName);
            if (null != version && version.length() > 0) {
                sn.append(":").append(version);
            }
            sn.append(" ");
            sn.append(inv.getMethodName());
            sn.append("(");
            Class<?>[] types = inv.getParameterTypes();
            if (types != null && types.length > 0) {
                boolean first = true;
                for (Class<?> type : types) {
                    if (first) {
                        first = false;
                    } else {
                        sn.append(",");
                    }
                    sn.append(type.getName());
                }
            }
            sn.append(") ");
            Object[] args = inv.getArguments();
            if (args != null && args.length > 0) {
                sn.append(new Gson().toJson(args));
            }
            String msg = sn.toString();
            log("accesslog", msg);
        } catch (Throwable t) {
            logger.warn("Exception in AcessLogFilter of service(" + invoker + " -> " + inv + ")", t);
        }
        return invoker.invoke(inv);
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/ClientChannelEventListener.java<|end_filename|>
package com.colobu.rpcx.netty;

import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author <EMAIL>
 */
public class ClientChannelEventListener implements ChannelEventListener {

    private static final Logger logger = LoggerFactory.getLogger(ClientChannelEventListener.class);


    public ClientChannelEventListener() {
    }

    @Override
    public void onChannelConnect(String remoteAddr, Channel channel) {
        logger.info("client connect remoteAddr:{}", remoteAddr);
    }

    @Override
    public void onChannelClose(String remoteAddr, Channel channel) {
        logger.info("client close remoteAddr:" + remoteAddr);
    }

    @Override
    public void onChannelException(String remoteAddr, Channel channel) {
        logger.info("client onChannelException  remoteAddr:{}", remoteAddr);
    }

    @Override
    public void onChannelIdle(String remoteAddr, Channel channel) {

    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/ExceptionFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.*;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.rpc.impl.RpcResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author <EMAIL>
 */
@RpcFilter(group = {Constants.PROVIDER}, order = -1999)
public class ExceptionFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(ExceptionFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        try {
            Result result = invoker.invoke(invocation);
            return result;
        } catch (RpcException ex) {
            Result result = new RpcResult();
            result.setThrowable(ex);
            return result;
        } catch (Throwable throwable) {
            Result result = new RpcResult();
            result.setThrowable(throwable);
            return result;
        }
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/RemotingCommand.java<|end_filename|>
package com.colobu.rpcx.protocol;

import com.colobu.rpcx.config.Constants;

/**
 * Created by <EMAIL>.
 */
public class RemotingCommand {


    private int code = 0;
    private int version = 0;


    /**
     * 解码的时候会用到,不会实际传输
     */
    private transient byte[] data;

    private Message message;

    public Message getMessage() {
        return message;
    }

    public void setMessage(Message message) {
        this.message = message;
        this.setOpaque((int) this.message.getSeq());
    }

    public RemotingCommand() {
    }

    public RemotingCommand(Message message, byte[] body) {
        this.message = message;
        this.data = body;
    }


    public static RemotingCommand createRequestCommand(Message message) {
        RemotingCommand cmd = new RemotingCommand();
        cmd.setMessage(message);
        return cmd;
    }


    public static RemotingCommand createResponseCommand(Message message) {
        RemotingCommand cmd = new RemotingCommand();
        cmd.setMessage(message);
        return cmd;
    }



    public void markResponseType() {
        this.message.setMessageType(MessageType.Response);
    }


    public int getCode() {
        return code;
    }

    public void setCode(int code) {
        this.code = code;
    }

    public int getVersion() {
        return version;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public int getOpaque() {
        return (int) this.message.getSeq();
    }

    public void setOpaque(int opaque) {
        this.message.setSeq(opaque);
    }


    public boolean isOnewayRPC() {
        return this.message.isOneway();
    }


    public RemotingCommandType getType() {
        if (this.isResponseType()) {
            return RemotingCommandType.RESPONSE_COMMAND;
        }
        return RemotingCommandType.REQUEST_COMMAND;
    }


    public boolean isResponseType() {
        return this.message.getMessageType().equals(MessageType.Response);
    }

    public void markOnewayRPC() {
        this.message.setOneway(true);
    }

    public static RemotingCommand createResponseCommand(int errorCode, String errorMessage) {
        RemotingCommand cmd = new RemotingCommand();
        cmd.markResponseType();
        cmd.getMessage().setMessageStatusType(MessageStatusType.Error);
        cmd.getMessage().metadata.put(Constants.RPCX_ERROR_CODE, String.valueOf(errorCode));
        cmd.getMessage().metadata.put(Constants.RPCX_ERROR_MESSAGE, errorMessage);
        return cmd;
    }

    public byte[] getData() {
        return data;
    }

    public void setData(byte[] data) {
        this.data = data;
    }


    public void setErrorMessage(int code, String message) {
        this.setErrorMessage(String.valueOf(code), message);
    }


    /**
     * rpcx 是通过 meta 传递错误信息的
     *
     * @param code
     * @param message
     */
    public void setErrorMessage(String code, String message) {
        //带有错误的返回结果
        this.message.setMessageStatusType(MessageStatusType.Error);
        this.message.metadata.put(Constants.RPCX_ERROR_CODE, code);
        this.message.metadata.put(Constants.RPCX_ERROR_MESSAGE, message);
    }

    public RemotingCommand requestToResponse() {
        this.message.setMessageType(MessageType.Response);
        this.data = new byte[]{};
        this.message.payload = new byte[]{};
        this.message.metadata.clear();
        return this;
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyEncoder.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.common.RemotingHelper;
import com.colobu.rpcx.common.RemotingUtil;
import com.colobu.rpcx.protocol.RemotingCommand;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;

/**
 * @author <EMAIL>
 */
public class NettyEncoder extends MessageToByteEncoder<RemotingCommand> {

    private static final Logger log = LoggerFactory.getLogger(NettyEncoder.class);

    @Override
    public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out) {
        try {
            byte[] data = remotingCommand.getMessage().encode();
            out.writeBytes(data);
        } catch (Exception e) {
            log.error("encode exception, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
            if (remotingCommand != null) {
                log.error(remotingCommand.toString());
            }
            RemotingUtil.closeChannel(ctx.channel());
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/ClassUtils.java<|end_filename|>
package com.colobu.rpcx.common;

import com.colobu.rpcx.rpc.ReflectUtils;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author <EMAIL>
 */
public class ClassUtils {


    private ClassUtils() {

    }

    public static ClassLoader getDefaultClassLoader() {
        ClassLoader cl = null;
        try {
            cl = Thread.currentThread().getContextClassLoader();
        } catch (Throwable ex) {
            // Cannot access thread context ClassLoader - falling back...
        }
        if (cl == null) {
            // No thread context class loader -> use class loader of this class.
            cl = ClassUtils.class.getClassLoader();
            if (cl == null) {
                // getClassLoader() returning null indicates the bootstrap ClassLoader
                try {
                    cl = ClassLoader.getSystemClassLoader();
                } catch (Throwable ex) {
                    // Cannot access system ClassLoader - oh well, maybe the caller can live with null...
                }
            }
        }
        return cl;
    }


    public static Class getClassByName(String name) {
        Class<?> clazz = null;
        try {
            clazz = Class.forName(name);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return clazz;
    }

    public static String getMethodKey(String className, String methodName, String[] parameterNames) {
        return className + "." + methodName + "(" + Arrays.stream(parameterNames).collect(Collectors.joining(",")) + ")";
    }

    public static String[] getMethodParameterNames(Method method) {
        Class<?>[] types = method.getParameterTypes();
        return Stream.of(types).map(it -> ReflectUtils.getName(it)).toArray(String[]::new);
    }


    public static Method getMethod(String className, String methodName, String[] parameterTypeNames) {
        Class<?> clazz = ClassUtils.getClassByName(className);
        Class[] clazzArray = Stream.of(parameterTypeNames).map(it -> ReflectUtils.forName(it)).toArray(Class[]::new);
        Method m = null;
        try {
            m = clazz.getMethod(methodName, clazzArray);
        } catch (NoSuchMethodException e) {
            e.printStackTrace();
        }
        return m;
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/ChannelEventListener.java<|end_filename|>
package com.colobu.rpcx.netty;

import io.netty.channel.Channel;

/**
 * @author <EMAIL>
 */
public interface ChannelEventListener {

    void onChannelConnect(final String remoteAddr, final Channel channel);

    void onChannelClose(final String remoteAddr, final Channel channel);

    void onChannelException(final String remoteAddr, final Channel channel);

    void onChannelIdle(final String remoteAddr, final Channel channel);
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/utils/ZkClientTest.java<|end_filename|>
package com.colobu.rpcx.utils;

import org.junit.Test;

import java.util.HashSet;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by <EMAIL>.
 */
public class ZkClientTest {


    @Test
    public void testWatch() throws Exception {
//        ZkClient.ins().watch();
//
//        LinkedBlockingQueue<PathStatus> queue = new LinkedBlockingQueue<>();
//        new Thread(()->{
//           while(true) {
//               try {
//                   PathStatus ps = queue.take();
//                   System.out.println("---------->"+ps);
//               } catch (InterruptedException e) {
//                   e.printStackTrace();
//               }
//           }
//        }).start();
//
//        ZkClient.ins().watch(queue,"");
//        TimeUnit.HOURS.sleep(1);
    }


    @Test
    public void testCreate() throws Exception {
        ZkClient.ins().create("",new HashSet<>(),"");
        TimeUnit.HOURS.sleep(1);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/LanguageCode.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum LanguageCode {
    /**
     * java
     */
    JAVA,
    /**
     * go
     */
    GO,
    /**
     * http
     */
    HTTP,
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/url/URLTest.java<|end_filename|>
package com.colobu.rpcx.url;

import com.colobu.rpcx.rpc.URL;
import org.junit.Test;

public class URLTest {


    @Test
    public void testUrl() {
        URL url = new URL("rpcx","",0);
        url.setPath("com.test.Service");
        url = url.addParameter("weight",22);
        url = url.addParameter("name","zzy");
        System.out.println(url.toFullString());

        System.out.println(url.getParameter("weight"));
        System.out.println(url.getPath());

        System.out.println(url.toParameterString());


        URL url2 = URL.valueOf("172.16.58.3:8888?a=3&b=2");
        System.out.println(url2.getAddress());
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/Invocation.java<|end_filename|>
package com.colobu.rpcx.rpc;

import java.util.Map;


/**
 * Created by <EMAIL>.
 */
public interface Invocation {

    String getMethodName();

    String getClassName();

    Class<?>[] getParameterTypes();

    String[] getParameterTypeNames();

    Object[] getArguments();

    Map<String, String> getAttachments();

    String getAttachment(String key);

    String getAttachment(String key, String defaultValue);

    Invoker<?> getInvoker();

    Class<?> getResultType();

    void setArguments(Object[] arguments);

    void setParameterTypeNames(String[] parameterTypeNames);


}


<|start_filename|>server/src/main/java/com/colobu/rpcx/processor/RpcHttpProcessor.java<|end_filename|>
package com.colobu.rpcx.processor;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.netty.NettyRequestProcessor;
import com.colobu.rpcx.protocol.*;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.ReflectUtils;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.impl.Exporter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.gson.Gson;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.function.Function;
import java.util.stream.IntStream;

import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

/**
 * @author <EMAIL>
 * 处理gateway过来的请求
 */
public class RpcHttpProcessor implements NettyRequestProcessor {

    private static final Logger logger = LoggerFactory.getLogger(RpcHttpProcessor.class);

    private final Function<Class, Object> getBeanFunc;

    public RpcHttpProcessor(Function<Class, Object> getBeanFunc) {
        this.getBeanFunc = getBeanFunc;
    }

    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) {
        Message req = request.getMessage();
        Gson gson = new Gson();
        RpcInvocation invocation = gson.fromJson(new String(request.getMessage().payload), RpcInvocation.class);
        String[] names = invocation.getParameterTypeNames();
        Object[] args = invocation.getArguments();
        invocation.setArguments(IntStream.range(0, names.length).mapToObj(i -> {
            String name = names[i];
            Class<?> clazz = ReflectUtils.forName(name);
            return gson.fromJson(args[i].toString(), clazz);
        }).toArray());
        invocation.languageCode = LanguageCode.HTTP;

        invocation.setClassName(request.getMessage().servicePath);
        invocation.setMethodName(request.getMessage().serviceMethod);
        invocation.setUrl(new URL("rpcx", "", 0));
        invocation.url.setHost(req.metadata.get("_host"));
        invocation.url.setPort(Integer.parseInt(req.metadata.get("_port")));
        invocation.getAttachments().put(Constants.TRACE_ID,req.metadata.get(Constants.X_RPCX_TRACEID));


        Message resMessage = new Message();
        resMessage.servicePath = invocation.getClassName();
        resMessage.serviceMethod = invocation.getMethodName();
        resMessage.setMessageType(MessageType.Response);
        resMessage.setSeq(request.getOpaque());


        String key = ClassUtils.getMethodKey(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
        Invoker<Object> wrapperInvoker = Exporter.invokerMap.get(key);


        //提供的class+method+paramType 找不到 provider
        if (null == wrapperInvoker) {
            resMessage.metadata.put(Constants.RPCX_ERROR_CODE, String.valueOf(RemotingSysResponseCode.SYSTEM_ERROR));
            resMessage.metadata.put(Constants.RPCX_ERROR_MESSAGE, "don't find service:" + key);
            resMessage.setPayload(new byte[]{});
            byte[] data = new Gson().toJson(resMessage).getBytes();
            sendResponse(ctx, data);
            return RemotingCommand.createResponseCommand(resMessage);
        }


        Result rpcResult = wrapperInvoker.invoke(invocation);
        resMessage.metadata.put(Constants.TRACE_ID, rpcResult.getAttachment(Constants.TRACE_ID));
        resMessage.metadata.put(Constants.SPAN_ID, rpcResult.getAttachment(Constants.SPAN_ID));
        resMessage.payload = new Gson().toJson(rpcResult.getValue()).getBytes();


        if (rpcResult.hasException()) {
            logger.error(rpcResult.getException().getMessage(), rpcResult.getException());
            resMessage.metadata.put(Constants.RPCX_ERROR_CODE, String.valueOf(RemotingSysResponseCode.SYSTEM_ERROR));
            resMessage.metadata.put(Constants.RPCX_ERROR_MESSAGE, rpcResult.getException().getMessage());
            resMessage.setPayload(new byte[]{});
        }

        RemotingCommand res = RemotingCommand.createResponseCommand(resMessage);

        logger.info("message:{}", res.getMessage());

        byte[] data = new Gson().toJson(res.getMessage()).getBytes();

        sendResponse(ctx, data);

        return res;
    }


    private void sendResponse(ChannelHandlerContext ctx, byte[] data) {
        FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1,
                OK, Unpooled.wrappedBuffer(data));
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH,
                response.content().readableBytes());
        response.headers().set(HttpHeaderNames.CONNECTION, "close");
        ctx.write(response);
        ctx.flush();
    }

    @Override
    public boolean rejectRequest() {
        return false;
    }
}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/Callable.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.protocol.Message;

/**
 * Created by <EMAIL>.
 */
public interface Callable {
    void call(Message msg);
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/Result.java<|end_filename|>
package com.colobu.rpcx.rpc;

import java.util.Map;


/**
 * Created by <EMAIL>.
 */
public interface Result {

    void setValue(Object value);

    Object getValue();

    Throwable getException();

    boolean hasException();

    Map<String, String> getAttachments();

    String getAttachment(String key);

    String getAttachment(String key, String defaultValue);

    void setThrowable(Throwable throwable);

}


<|start_filename|>common/src/test/java/com/colobu/rpcx/common/YamlTest.java<|end_filename|>
package com.colobu.rpcx.common;

import com.colobu.rpcx.config.RpcxConfig;
import org.junit.Test;
import org.yaml.snakeyaml.DumperOptions;
import org.yaml.snakeyaml.Yaml;

import java.util.Map;

public class YamlTest {


    @Test
    public void testYaml() {
        String yamlStr = "key: hello yaml";
        Yaml yaml = new Yaml();
        Map<String, String> ret = yaml.load(yamlStr);
        System.out.println(ret);
        System.out.println(ret.get("key"));
    }


    @Test
    public void testDump() {
        System.out.println(dump());
    }

    private String dump() {
        DumperOptions options = new DumperOptions();
        options.setPrettyFlow(true);
        Yaml yaml = new Yaml(options);
        RpcxConfig config = new RpcxConfig();
        config.setConsumerPackage("com.test.consumer");
        config.setFilterPackage("com.test.filter");
        return yaml.dump(config);
    }


    @Test
    public void testLoad() {
        Yaml yaml = new Yaml();
        RpcxConfig config = yaml.load(dump());
        System.out.println(config.getConsumerPackage());
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyRemotingAbstract.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.common.Config;
import com.colobu.rpcx.common.Pair;
import com.colobu.rpcx.common.RemotingHelper;
import com.colobu.rpcx.common.RemotingUtil;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.MessageType;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.protocol.RemotingSysResponseCode;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.epoll.Epoll;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author <EMAIL>
 */
public class NettyRemotingAbstract {

    private static final Logger logger = LoggerFactory.getLogger(NettyRemotingAbstract.class);

    protected final ConcurrentHashMap<Integer, ResponseFuture> responseTable = new ConcurrentHashMap<>(256);

    protected final HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>> processorTable = new HashMap<>(64);

    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;

    protected final NettyEventExecuter nettyEventExecuter = new NettyEventExecuter();

    protected final ChannelEventListener channelEventListener = new ClientChannelEventListener();

    protected final ConcurrentHashMap<String, ChannelWrapper> channelTables = new ConcurrentHashMap<>();

    protected final Lock lockChannelTables = new ReentrantLock();

    protected static final long LockTimeoutMillis = 3000;

    private LongAdder processingRequest = new LongAdder();

    private volatile boolean stop = false;

    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecuter.putNettyEvent(event);
    }


    public void scanProcessingRequest() {
        if (gracefullyShutDown) {
            logger.info("processing request num:{}", processingRequest.longValue());
        }
    }

    public void scanResponseTable() {
        final List<ResponseFuture> rfList = new LinkedList<>();
        //是协议对应号
        Iterator<Map.Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
                rep.release();
                it.remove();
                rfList.add(rep);
                logger.warn("remove timeout request:{} ", rep);
            }
        }

        for (ResponseFuture rf : rfList) {
            try {
                rf.executeInvokeCallback();
            } catch (Throwable e) {
                logger.warn("scanResponseTable, operationComplete Exception", e);
            }
        }
    }

    /**
     * 关闭channel
     *
     * @param channel
     */
    public void closeChannel(final Channel channel) {
        if (null == channel) {
            return;
        }
        try {
            if (this.lockChannelTables.tryLock(LockTimeoutMillis, TimeUnit.MILLISECONDS)) {
                try {
                    boolean removeItemFromTable = true;
                    ChannelWrapper prevCW = null;
                    String addrRemote = null;
                    for (Map.Entry<String, ChannelWrapper> entry : channelTables.entrySet()) {
                        String key = entry.getKey();
                        ChannelWrapper prev = entry.getValue();
                        if (prev.getChannel() != null) {
                            if (prev.getChannel() == channel) {
                                prevCW = prev;
                                addrRemote = key;
                                break;
                            }
                        }
                    }

                    if (null == prevCW) {
                        logger.info("eventCloseChannel: the channel[{}] has been removed from the channel table before", addrRemote);
                        removeItemFromTable = false;
                    }

                    if (removeItemFromTable) {
                        this.channelTables.remove(addrRemote);
                        logger.info("closeChannel: the channel[{}] was removed from channel table", addrRemote);
                        RemotingUtil.closeChannel(channel);
                    }
                } catch (Exception e) {
                    logger.error("closeChannel: close the channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                logger.warn("closeChannel: try to lock channel table, but timeout, {}ms", LockTimeoutMillis);
            }
        } catch (InterruptedException e) {
            logger.error("closeChannel exception", e);
        }
    }


    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) {
        final RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

    private static final boolean gracefullyShutDown = Boolean.parseBoolean(Config.ins().get("rpcx.gracefully.shutdown", "false"));

    public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        final int opaque = cmd.getOpaque();
        if (this.stop) {
            if (!cmd.isOnewayRPC()) {
                Message message = new Message();
                message.setMessageType(MessageType.Response);
                message.setSeq(opaque);
                message.metadata.put(Constants.RPCX_ERROR_CODE, "-9");
                message.metadata.put(Constants.RPCX_ERROR_MESSAGE, "server begin shutdown");
                final RemotingCommand response = RemotingCommand.createResponseCommand(message);
                ctx.writeAndFlush(response);
            }
            return;
        }


        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;

        if (pair != null) {
            Runnable run = () -> {
                try {
                    if (gracefullyShutDown) {
                        processingRequest.increment();
                    }
                    final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
                    //是否是oneway是根据来的request计算的
                    if (!cmd.isOnewayRPC()) {
                        if (response != null) {
                            response.setOpaque(opaque);
                            response.markResponseType();
                            try {
                                ctx.writeAndFlush(response);
                            } catch (Throwable e) {
                                logger.error("process request over, but response failed", e);
                                logger.error(cmd.toString());
                                logger.error(response.toString());
                            }
                        }
                    }
                } catch (Throwable e) {
                    logger.error("process request exception", e);
                    logger.error(cmd.toString());
                    if (!cmd.isOnewayRPC()) {
                        final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR,
                                RemotingHelper.exceptionSimpleDesc(e));
                        response.setOpaque(opaque);
                        ctx.writeAndFlush(response);
                    }
                } finally {
                    if (gracefullyShutDown) {
                        processingRequest.decrement();
                    }
                }
            };

            if (pair.getObject1().rejectRequest()) {
                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                        "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }

            try {
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                pair.getObject2().submit(requestTask);

            } catch (RejectedExecutionException e) {
                if ((System.currentTimeMillis() % 10000) == 0) {
                    logger.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel())
                            + ", too many requests and system thread pool busy, RejectedExecutionException "
                            + pair.getObject2().toString()
                            + " request code: " + cmd.getCode());
                }

                if (!cmd.isOnewayRPC()) {
                    final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                            "[OVERLOAD]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                }
            }
        } else {
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            logger.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }
    }

    protected boolean useEpoll() {
        return RemotingUtil.isLinuxPlatform() && Epoll.isAvailable();
    }


    /**
     * 处理返回结果
     *
     * @param ctx
     * @param cmd
     */
    public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        try {
            cmd.getMessage().decode(cmd.getData());
        } catch (UnsupportedEncodingException e) {
            logger.warn("decode error:{}", e.getMessage());
        }
        //获取request带过去的唯一码
        final int opaque = cmd.getOpaque();
        //查询responseFuture
        final ResponseFuture responseFuture = this.responseTable.get(opaque);
        if (responseFuture != null) {
            //这里并不解除阻塞
            responseFuture.setResponseCommand(cmd);
            responseFuture.release();
            this.responseTable.remove(opaque);
            if (responseFuture.getInvokeCallback() != null) {
                boolean runInThisThread = false;
                ExecutorService executor = this.getCallbackExecutor();
                if (executor != null) {
                    try {
                        executor.submit(() -> {
                            try {
                                responseFuture.executeInvokeCallback();
                            } catch (Throwable e) {
                                logger.warn("execute callback in executor exception, and callback throw", e);
                            }
                        });
                    } catch (Exception e) {
                        runInThisThread = true;
                        logger.warn("execute callback in executor exception, maybe executor busy", e);
                    }
                } else {
                    runInThisThread = true;
                }

                if (runInThisThread) {
                    try {
                        responseFuture.executeInvokeCallback();
                    } catch (Throwable e) {
                        logger.warn("executeInvokeCallback Exception", e);
                    }
                }
            } else {
                //解除阻塞
                responseFuture.putResponse(cmd);
            }
        } else {
            logger.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            logger.warn(cmd.toString());
        }
    }

    private ExecutorService callbackExecutor = Executors.newFixedThreadPool(5);

    private ExecutorService getCallbackExecutor() {
        return callbackExecutor;
    }

    public boolean hasEventListener() {
        return this.channelEventListener != null;
    }

    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executor);
        this.processorTable.put(requestCode, pair);
    }

    public void serverShutdown() {
        logger.info("server shutdown begin");
        stop = true;
        int i = 0;
        if (gracefullyShutDown) {
            while (true) {
                if (i++ >= 30) {
                    break;
                }
                long num = this.processingRequest.sum();
                logger.info("shutdown processing request num:{}", num);
                if (num == 0) {
                    break;
                }
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        logger.info("server shutdown finish");
    }

}


<|start_filename|>container/src/main/java/com/colobu/rpcx/spring/RpcxAutoConfigure.java<|end_filename|>
package com.colobu.rpcx.spring;

import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.client.NettyClient;
import com.colobu.rpcx.client.ZkServiceDiscovery;
import com.colobu.rpcx.netty.IClient;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.Provider;
import com.colobu.rpcx.register.IServiceRegister;
import com.colobu.rpcx.server.NettyServer;
import com.colobu.rpcx.server.ZkServiceRegister;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.reflections.Reflections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.Set;
import java.util.UUID;
import java.util.function.Function;

/**
 * @author <EMAIL>
 */
@Configuration
@Aspect
@ConditionalOnClass(RpcxConsumer.class)
public class RpcxAutoConfigure {

    private static final Logger logger = LoggerFactory.getLogger(RpcxAutoConfigure.class);

    @Autowired
    private ApplicationContext context;

    @Value("${rpcx.provider.package.path}")
    private String rpcxProviderPackagePath;

    @Value("${rpcx.consumer.package.path}")
    private String rpcxConsumerPackagePath;

    @Value("${rpcx.base.path}")
    private String rpcxBasePath;


    @PostConstruct
    private void init() {
        Reflections reflections = new Reflections(rpcxProviderPackagePath);
        Set<Class<?>> providerSet = reflections.getTypesAnnotatedWith(Provider.class, true);
        providerSet.stream().forEach(it -> {
            logger.info("provider:{}", it);
        });

        NettyServer server = new NettyServer();
        Function<Class, Object> getBeanFunc = (clazz) -> context.getBean(clazz);
        server.setGetBeanFunc(getBeanFunc);
        server.start();
        IServiceRegister reg = new ZkServiceRegister(rpcxBasePath, server.getAddr() + ":" + server.getPort(), rpcxProviderPackagePath, getBeanFunc);
        reg.register();
        reg.start();
    }

    @Bean(destroyMethod = "close")
    @ConditionalOnMissingBean
    public RpcxConsumer rpcxConsumer() {
        IServiceDiscovery serviceDiscovery = new ZkServiceDiscovery(rpcxBasePath, rpcxConsumerPackagePath);
        IClient client = new NettyClient(serviceDiscovery);
        return new RpcxConsumer(client);
    }


    /**
     * 所有provider 的执行,这里都会被拦截到
     *
     * @param joinPoint
     * @return
     */
//    @Around("execution(public * *(..)) && within(@com.colobu.rpcx.rpc.annotation.Provider *)")
    public Object aroundProvider(ProceedingJoinPoint joinPoint) throws Throwable {
        Provider provider = joinPoint.getTarget().getClass().getAnnotation(Provider.class);
        String uuid = UUID.randomUUID().toString();
        Object[] o = joinPoint.getArgs();
        logger.debug("provider execute begin name:{} version:{}  id:{} params:{}", provider.name(), provider.version(), uuid, joinPoint.getArgs());
        long startTime = System.currentTimeMillis();
        try {
            Object result = joinPoint.proceed();
            long useTime = System.currentTimeMillis() - startTime;
            logger.debug("provider execute finish name:{} version:{} id:{} result:{} useTime:{}", provider.name(), provider.version(), uuid, result, useTime);
            return result;
        } catch (RpcException ex) {
            logger.warn("provider execute finish name:{} version:{} id:{} error:{}", provider.name(), provider.version(), uuid, ex.getMessage());
            throw ex;
        } catch (Throwable throwable) {
            logger.warn("provider execute finish name:{} version:{} id:{} error:{}", provider.name(), provider.version(), uuid, throwable.getMessage());
            throw throwable;
        }
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/deploy/AgentLoader.java<|end_filename|>
package com.colobu.rpcx.deploy;

import com.sun.tools.attach.VirtualMachine;

import java.io.IOException;

public class AgentLoader {


    public void loadAgent(String processId, String jarFileName, String params) {
        VirtualMachine virtualMachine = null;
        try {
            virtualMachine = VirtualMachine.attach(processId);
            virtualMachine.loadAgent(jarFileName, params);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                virtualMachine.detach();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/Config.java<|end_filename|>
package com.colobu.rpcx.common;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;

import java.util.Map;
import java.util.Properties;

/**
 * @author <EMAIL>
 */
public class Config {

    private static final Logger logger = LoggerFactory.getLogger(Config.class);

    private Properties properties;

    private Config() {
        properties = new ClassPathResource("application.properties").getProperties();
        //拥有多个配置文件
        if (properties.containsKey("spring.profiles.active")) {
            Properties propertiesTmp = new ClassPathResource("application-" + properties.get("spring.profiles.active") + ".properties").getProperties();
            propertiesTmp.keySet().stream().forEach(key -> properties.put(key, propertiesTmp.get(key)));
        }
        String yamlProperties = new ClassPathResource("application.yaml").getString();
        if (StringUtils.isNotEmpty(yamlProperties)) {
            Yaml yaml = new Yaml();
            Map<String, String> m = yaml.load(yamlProperties);
            m.entrySet().stream().forEach(kv -> {
                if (kv.getKey().startsWith("rpcx.")) {
                    properties.put(kv.getKey(), String.valueOf(kv.getValue()));
                }
            });
        }
        logger.info("read config finish :{}", this.properties.keySet());
    }


    private static class LazyHolder {
        private static final Config ins = new Config();
    }


    public static Config ins() {
        return LazyHolder.ins;
    }

    public String get(String key) {
        return this.properties.getProperty(key);
    }

    public String get(String key, String defaultValue) {
        return this.properties.getProperty(key, defaultValue);
    }
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/common/CommonTest.java<|end_filename|>
package com.colobu.rpcx.common;

import com.colobu.rpcx.common.retry.RetryNTimes;
import com.colobu.rpcx.common.retry.RetryPolicy;
import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.rpc.A;
import com.colobu.rpcx.rpc.HessianUtils;
import com.colobu.rpcx.rpc.ReflectUtils;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.esotericsoftware.reflectasm.MethodAccess;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.junit.Test;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * Created by <EMAIL>.
 */
public class CommonTest {


    @Test
    public void testString() {
        String str = "abcde";
        byte[] b = str.getBytes();

        System.out.println(b[3]);

        System.out.println(b.length);

        String str2 = new String(b, 1, b.length - 1);
        b[1] = 90;
        System.out.println(str2);

        System.out.println("------------->");

        long begin = System.currentTimeMillis();
        IntStream.range(0, 10000000).forEach(it -> {
            byte[] b2 = new byte[b.length-1];
            System.arraycopy(b, 1, b2, 0, b.length-1);
            String st = new String(b2);
//            System.out.println(st);
        });
        System.out.println(System.currentTimeMillis() - begin);

        begin = System.currentTimeMillis();
        IntStream.range(0, 10000000).forEach(it -> {
            String st = new String(b, 1, b.length-1);
//            System.out.println(st);
        });
        System.out.println(System.currentTimeMillis() - begin);
    }


    @Test
    public void testFlag() {
//        RemotingCommand c = RemotingCommand.createResponseCommand();
        RemotingCommand c = RemotingCommand.createRequestCommand(new Message());
        int bits = 1 << 0;
        System.out.println(bits);

    }


    @Test
    public void testEncode1() throws IOException {

        Map<String, String> metadata = Maps.newHashMap();
        metadata.put("a", "11111111111111111111111");
        metadata.put("b", "21111111111111111111111");
        metadata.put("c", "31111111111111111111111");

        long begin = System.currentTimeMillis();

        ByteArrayOutputStream os = new ByteArrayOutputStream(20);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes("UTF-8");
            os.write(Bytes.int2bytes(keyBytes.length));
            os.write(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes("UTF-8");
            os.write(Bytes.int2bytes(vBytes.length));
            os.write(vBytes);
        }

        byte[] d = os.toByteArray();
        System.out.println(System.currentTimeMillis() - begin);
        System.out.println(Arrays.toString(d));
    }

    @Test
    public void testEncode2() throws IOException {

        Map<String, String> metadata = Maps.newHashMap();
        metadata.put("a", "11111111111111111111111");
        metadata.put("b", "21111111111111111111111");
        metadata.put("c", "31111111111111111111111");

        long begin = System.currentTimeMillis();

        ByteArrayOutputStream os = new ByteArrayOutputStream(20);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes("UTF-8");
            os.write(Bytes.int2bytes(keyBytes.length));
            os.write(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes("UTF-8");
            os.write(Bytes.int2bytes(vBytes.length));
            os.write(vBytes);
        }

        byte[] d = os.toByteArray();
        System.out.println(System.currentTimeMillis() - begin);
        System.out.println(Arrays.toString(d));
    }

    @Test
    public void testEncode3() throws IOException {

        Map<String, String> metadata = Maps.newHashMap();
        metadata.put("a", "11111111111111111111111");
        metadata.put("b", "21111111111111111111111");
        metadata.put("c", "31111111111111111111111");

        long begin = System.currentTimeMillis();

        ByteBuf buffer = Unpooled.buffer(metadata.size() * 30);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes();

            buffer.writeBytes(Bytes.int2bytes(keyBytes.length));
            buffer.writeBytes(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes();
            buffer.writeBytes(Bytes.int2bytes(vBytes.length));
            buffer.writeBytes(vBytes);
        }

        buffer.capacity(buffer.writerIndex());
        System.out.println(System.currentTimeMillis() - begin);
        System.out.println(Arrays.toString(buffer.array()));
    }


    //168
    @Test
    public void testDecode1() throws UnsupportedEncodingException {

        final byte[] b = new byte[]{0, 0, 0, 1, 97, 0, 0, 0, 23, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 0, 0, 0, 1, 98, 0, 0, 0, 23, 50, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 0, 0, 0, 1, 99, 0, 0, 0, 23, 51, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49};

        Map<String, String> m = Maps.newHashMap();
        long begin = System.currentTimeMillis();
        IntStream.range(0, 100000).forEach(it -> {
            ByteBuffer buf = ByteBuffer.wrap(b);
            int len;
            byte[] data = null;
            for (; ; ) {
                if (buf.remaining() < 4) {
                    break;
                }
                len = buf.getInt();
                data = new byte[len];
                buf.get(data);
                String k = null;
                try {
                    k = new String(data, "UTF-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }

                len = buf.getInt();
                data = new byte[len];
                buf.get(data);
                String v = null;
                try {
                    v = new String(data, "UTF-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                m.put(k, v);
            }
        });

        System.out.println(System.currentTimeMillis() - begin);
        System.out.println(m);
    }


    //148
    @Test
    public void testDecode2() throws UnsupportedEncodingException {

        byte[] d = new byte[]{0, 0, 0, 1, 97, 0, 0, 0, 23, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 0, 0, 0, 1, 98, 0, 0, 0, 23, 50, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 0, 0, 0, 1, 99, 0, 0, 0, 23, 51, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49, 49};
        Map<String, String> m = Maps.newHashMap();

        long begin = System.currentTimeMillis();
        IntStream.range(0, 100000).forEach(it -> {
            int blen = d.length;
            int index = 0;
            int len = 0;
            byte[] b = null;
            for (; ; ) {
                if (blen - index < 4) {
                    break;
                }
                len = Bytes.bytes2int(d, index);
                index = index + 4;
                b = new byte[len];
                System.arraycopy(d, index, b, 0, len);
                String k = null;
                try {
                    k = new String(b, "UTF-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                index = index + len;

                len = Bytes.bytes2int(d, index);
                index = index + 4;
                b = new byte[len];
                System.arraycopy(d, index, b, 0, len);
                index = index + len;
                String v = null;
                try {
                    v = new String(b, "UTF-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                m.put(k, v);
            }
        });

        System.out.println(System.currentTimeMillis() - begin);
        System.out.println(m);
    }


    //2540 20153
    @Test
    public void testMap2() {
        ExecutorService pool = Executors.newFixedThreadPool(20);

        ConcurrentHashMap<String, String> m2 = new ConcurrentHashMap<>();
        m2.put("a", "1");

        Stopwatch sw2 = Stopwatch.createStarted();
        IntStream.range(0, 1000000).forEach(i -> {
            try {
                pool.invokeAll(IntStream.range(0, 20).mapToObj(ii -> (Callable<Void>) () -> {
                    m2.get("a");
                    return null;
                }).collect(Collectors.toList()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(sw2.elapsed(TimeUnit.MILLISECONDS));

    }

    //2554 22076
    @Test
    public void testMap1() {
        ExecutorService pool = Executors.newFixedThreadPool(20);

        Map<String, String> m = new HashMap<>();
        m.put("a", "1");
        Stopwatch sw = Stopwatch.createStarted();
        IntStream.range(0, 1000000).forEach(i -> {
            try {
                pool.invokeAll(IntStream.range(0, 20).mapToObj(ii -> (Callable<Void>) () -> {
                    m.get("a");
                    return null;
                }).collect(Collectors.toList()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(sw.elapsed(TimeUnit.MILLISECONDS));


    }


    @Test
    public void testPool() throws Exception {
        Stopwatch sw = Stopwatch.createStarted();
        IntStream.range(0, 10000000).forEach(it -> {
            RemotingCommand.createRequestCommand(new Message());
        });
        System.out.println(sw.elapsed(TimeUnit.MILLISECONDS));

        TestPoolFactory factory = new TestPoolFactory();
        GenericObjectPool pool = new GenericObjectPool(factory);
        sw.reset();

        IntStream.range(0, 10000000).forEach(it -> {
            try {
                pool.addObject();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        sw.start();
        IntStream.range(0, 10000000).forEach(it -> {
            try {
                Object v = pool.borrowObject();
//                System.out.println(v);
                pool.returnObject(v);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        System.out.println(sw.elapsed(TimeUnit.MILLISECONDS));
    }


    @Test
    public void testPool2() {
        Stopwatch sw = Stopwatch.createStarted();
        IntStream.range(0, 10000000).forEach(it -> {
            RemotingCommand.createRequestCommand(new Message());
        });
        System.out.println(sw.elapsed(TimeUnit.MILLISECONDS));
    }


    @Test
    public void testBuffer() {
        byte[] b = new byte[]{0, 0, 0, 1, 0, 0, 0, 2};
        long begin = System.currentTimeMillis();
        IntStream.range(0, 1).forEach(it -> {
            ByteBuffer buffer = ByteBuffer.wrap(b);
            buffer.position(4);
            System.out.println(buffer.getInt());
        });
        System.out.println(System.currentTimeMillis() - begin);

        begin = System.currentTimeMillis();
        IntStream.range(0, 1).forEach(it -> {
            System.out.println(Bytes.bytes2int(b, 4));
        });
        System.out.println(System.currentTimeMillis() - begin);

    }


    @Test
    public void testWrite() {

        Map<String, String> m = Maps.newHashMap();
        m.put("a", "aaaaaaaaaaaaaaaaaad");
        m.put("b", "aaaaaaaaaaaaaaaaaad");

        long now = System.currentTimeMillis();
        IntStream.range(0, 1).forEach(i -> {
            try {
                System.out.println(Arrays.toString(encodeMetadata(m)));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        System.out.println(System.currentTimeMillis() - now);

        now = System.currentTimeMillis();

        IntStream.range(0, 1).forEach(i -> {
            try {
                System.out.println(Arrays.toString(encodeMetadata2(m)));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        System.out.println(System.currentTimeMillis() - now);
    }


    private byte[] encodeMetadata(Map<String, String> metadata) throws IOException {
        if (metadata.size() == 0) {
            return new byte[]{};
        }
        ByteArrayOutputStream os = new ByteArrayOutputStream(20);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes("UTF-8");
            os.write(ByteBuffer.allocate(4).putInt(keyBytes.length).array());
            os.write(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes("UTF-8");
            os.write(ByteBuffer.allocate(4).putInt(vBytes.length).array());
            os.write(vBytes);
        }

        return os.toByteArray();
    }

    private byte[] encodeMetadata2(Map<String, String> metadata) throws IOException {
        if (metadata.size() == 0) {
            return new byte[]{};
        }

        ByteBuf buffer = Unpooled.buffer(20);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes("UTF-8");
            buffer.writeInt(keyBytes.length);
            buffer.writeBytes(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes("UTF-8");
            buffer.writeInt(vBytes.length);
            buffer.writeBytes(vBytes);
        }


        return buffer.array();
    }


class Bean {
    public String str(String s) {
        return "str:" + s;
    }
}


    @Test
    public void testMethdHandle() throws Throwable {
        MethodHandles.Lookup lookup = MethodHandles.lookup();
        MethodType mt = MethodType.methodType(String.class, String.class);
        MethodHandle mh = lookup.findVirtual(Bean.class, "str", mt);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 10000000; i++) {
            String s = (String) mh.invokeExact(new Bean(), "zzy");
//            System.out.println(s);
        }
        System.out.println(System.currentTimeMillis() - begin);
    }

    //1702 1562 1537
    @Test
    public void testReflect() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
        Bean bean = new Bean();
        Method method = bean.getClass().getMethod("str", String.class);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 100000000; i++) {
            String res = (String) method.invoke(bean, "zzy");
//            System.out.println(res);
        }
        System.out.println(System.currentTimeMillis() - begin);
    }

    //1340 1244 1202
    @Test
    public void testReflect2() {
        Bean bean = new Bean();
        MethodAccess access = MethodAccess.get(Bean.class);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < 100000000; i++) {
            String r = (String) access.invoke(bean, "str", "zzy");
//            System.out.println(r);
        }
        System.out.println(System.currentTimeMillis() - begin);
    }


//    @Test
//    public void testReflectAsm() {
//        Bean bean = new Bean();
//        MethodAccess access = MethodAccess.get(Bean.class);
//        long begin = System.currentTimeMillis();
//        for (int i = 0; i < 100000000; i++) {
//            String res = (String) access.invoke(bean, "str", "zzy");
////            System.out.println(res);
//        }
//        System.out.println(System.currentTimeMillis() - begin);
//    }


    @Test
    public void testClassName() {
        System.out.println(ReflectUtils.getName(byte[].class));
    }


    @Test
    public void testCode() {
        RemotingCommand c = RemotingCommand.createResponseCommand(new Message());
        c.markOnewayRPC();
        System.out.println(c.isOnewayRPC());
    }


    @Test
    public void testClassUtils() throws ClassNotFoundException {

        String desc = ReflectUtils.getDesc(int.class);
        System.out.println(desc);

        System.out.println(ReflectUtils.desc2name(desc));


        System.out.println(ReflectUtils.getName(int.class));

        System.out.println(ReflectUtils.name2class("int"));

//        System.out.println(ClassUtils.getClassByName("int.class"));
    }


    @Test
    public void getDesc() {
        String desc = ReflectUtils.getDesc(new byte[]{}.getClass());
        System.out.println(desc);
    }


    @Test
    public void testRetry() {
        RetryPolicy retryPolicy = new RetryNTimes(10);
        retryPolicy.retry((i) -> {
            System.out.println(i);
            if (i == 3) {
            }
            return null;
        });
    }


    @Test
    public void testConfig() {
        System.out.println(Config.ins().get("server.type"));
        System.out.println(Config.ins().get("server.type1", "aaaa"));
    }


    @Test
    public void testClassPathResource() {
        System.out.println(new ClassPathResource("test").getString());
        System.out.println(new ClassPathResource("test").getProperties().getProperty("2"));
    }


class E {
    public E() {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            System.out.println("schedule");
        }, 10, 1, TimeUnit.SECONDS);

        Executors.newSingleThreadScheduledExecutor()
                .scheduleWithFixedDelay(() -> {
                    System.out.println("schedule");
                }, 0, 5, TimeUnit.SECONDS);
    }

}


    @Test
    public void testSchedule() {
        E e = new E();
        System.out.println(e);
        System.out.println("finish");
    }


    @Test
    public void testSwitch() {
        int i = 12;
        switch (i) {
            case 12:
                System.out.println(12);
                break;
            case 13:
                System.out.println(13);
                break;
            default:
                System.out.println(14);
        }
    }


    @Test
    public void testLocalAddr() {
        System.out.println(RemotingUtil.getLocalAddress());
    }


    @Test
    public void testClazz() {
        int i = 12;
        System.out.println(new Object[]{i}[0].getClass().getTypeName());


        System.out.println(int.class.getName());
    }


    @Test
    public void testGson() {
        Gson gson = new GsonBuilder().excludeFieldsWithoutExposeAnnotation().create();
        RpcInvocation invocation = new RpcInvocation();
        invocation.setClassName("name");
        String data = gson.toJson(invocation);
        System.out.println(data);


        int i = 12;
        String str = new Gson().toJson(i);
        System.out.println(str);

        int i2 = new Gson().fromJson(str, int.class);
        System.out.println(i2);
    }


    @Test
    public void testHessian() {
        Object[] obj = new Object[]{1, 2, 3, 4, "a", 1.0, new A()};
        byte[] data = HessianUtils.write(obj);
        System.out.println(Arrays.toString(data));

        Object[] obj2 = (Object[]) HessianUtils.read(data);
        System.out.println(Arrays.toString(obj2));

    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/HotDeployFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.common.Config;
import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.deploy.AgentLoader;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.rpc.impl.RpcResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.UUID;

/**
 * @author <EMAIL>
 * 支持热更新
 */
@RpcFilter(order = -2002, group = {Constants.PROVIDER})
public class HotDeployFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(HotDeployFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation inv) throws RpcException {
        if (inv.getMethodName().equals(Constants.$HOT_DEPLOY) && inv.getArguments().length == 3) {
            logger.info("hot deploy begin");
            Object[] arguments = inv.getArguments();

            String className = arguments[0].toString();
            String token = arguments[1].toString();

            String configToken = Config.ins().get("rpcx.deploy.tokey");
            if (StringUtils.isEmpty(configToken) || !token.equals(configToken)) {
                throw new RpcException("token error");
            }

            byte[] classData = (byte[]) arguments[2];
            String uuid = UUID.randomUUID().toString();
            String file = "/tmp/" + uuid;
            String pid = getPid();
            logger.info("className:{} file:{} pid:{} data len:{}", className, file, pid, classData.length);
            try {
                Files.write(Paths.get(file), classData);
            } catch (IOException e) {
                e.printStackTrace();
            }
            new AgentLoader().loadAgent(pid, Config.ins().get("rpcx_agent_path"), file + "__" + className);

            try {
                Files.delete(Paths.get(file));
            } catch (IOException e) {
                e.printStackTrace();
            }
            return new RpcResult("Finish");
        }

        return invoker.invoke(inv);
    }


    public String getPid() {
        RuntimeMXBean runtime = ManagementFactory.getRuntimeMXBean();
        // format: "pid@hostname"
        String name = runtime.getName();
        return name.substring(0, name.indexOf('@'));
    }

}

<|start_filename|>client/src/test/java/com/colobu/rpcx/client/ServiceDiscoveryTest.java<|end_filename|>
package com.colobu.rpcx.client;

import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * Created by zhangzhiyong on 2018/7/4.
 */
public class ServiceDiscoveryTest {


    @Test
    public void testWatch() throws Exception {
        new ZkServiceDiscovery("","").watch();
        TimeUnit.HOURS.sleep(1);
    }
}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/ServiceAddr.java<|end_filename|>
package com.colobu.rpcx.client;

/**
 * Created by <EMAIL>.
 */
public class ServiceAddr {


}


<|start_filename|>common/src/test/java/com/colobu/rpcx/rpc/A.java<|end_filename|>
package com.colobu.rpcx.rpc;

import java.io.Serializable;

public class A implements Serializable {
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/RpcProviderInvoker.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.rpc.*;
import com.esotericsoftware.reflectasm.MethodAccess;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.function.Function;

/**
 * @author <EMAIL>
 */
public class RpcProviderInvoker<T> implements Invoker<T> {

    private static final Logger logger = LoggerFactory.getLogger(RpcProviderInvoker.class);

    private static final boolean useMethodAccess = true;

    /**
     * 这个invoker 是否被共享(泛化调用就是共享的)
     * 如果是共享的 则class 和 method 都是不能设值的,不然并发下是有问题的
     */
    private boolean share = false;

    /**
     * 如果是基于ioc容器的,需要提供获取bean的function
     */
    private Function<Class, Object> getBeanFunc;

    private URL url;

    private Class clazz;

    private Method method;

    private MethodAccess methodAccess;

    public RpcProviderInvoker(Function<Class, Object> getBeanFunc) {
        this.getBeanFunc = getBeanFunc;
    }

    @Override
    public Class<T> getInterface() {
        return clazz;
    }


    @Override
    public Result invoke(RpcInvocation invocation) {
        Result rpcResult = new RpcResult();
        String traceId = "";
        String spanId = "";
        try {
            Object obj = null;
            traceId = RpcContext.getContext().getAttachments().get(Constants.TRACE_ID);
            spanId = RpcContext.getContext().getAttachments().get(Constants.SPAN_ID);
            //设置traceId
            rpcResult.getAttachments().put(Constants.TRACE_ID, traceId == null ? "" : traceId);
            rpcResult.getAttachments().put(Constants.SPAN_ID, spanId == null ? "" : spanId);
            //使用容器
            if (null != this.getBeanFunc) {
                if (share) {
                    //泛化调用,每次都创建不同的
                    Class c = ClassUtils.getClassByName(invocation.getClassName());
                    Object b = getBeanFunc.apply(c);
                    Method method = ClassUtils.getMethod(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
                    obj = method.invoke(b, invocation.getArguments());
                } else {
                    //每次都调用相同的,应为class 和 method 都已经实例化好了
                    Object b = getBeanFunc.apply(clazz);
                    if (useMethodAccess) {
                        obj = methodAccess.invoke(b, invocation.getMethodName(), invocation.getArguments());
                    } else {
                        obj = this.method.invoke(b, invocation.getArguments());
                    }
                }
            } else {//不使用容器
                if (share) {
                    Class c = ClassUtils.getClassByName(invocation.getClassName());
                    Object b = getBeanFunc.apply(c);
                    Method method = ClassUtils.getMethod(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
                    obj = method.invoke(b, invocation.getArguments());
                } else {
                    if (useMethodAccess) {
                        obj = methodAccess.invoke(clazz.newInstance(), invocation.getMethodName(), invocation.getArguments());
                    } else {
                        obj = this.method.invoke(clazz.newInstance(), invocation.getArguments());
                    }
                }
            }
            rpcResult.setValue(obj);
            return rpcResult;
        } catch (Throwable throwable) {
            String message = throwable.getMessage();
            if (throwable.getCause() != null) {
                message = throwable.getCause().getMessage();
            }
            message = "traceId:" + traceId + ", message:" + message;
            RpcException ex = new RpcException(message, throwable, "-2");
            rpcResult.setThrowable(ex);
            return rpcResult;
        }
    }

    @Override
    public URL getUrl() {
        return this.url;
    }

    @Override
    public boolean isAvailable() {
        return true;
    }

    @Override
    public void destroy() {

    }

    @Override
    public void setMethod(Method method) {
        this.method = method;
    }

    @Override
    public Method getMethod() {
        return this.method;
    }

    @Override
    public void setInterface(Class clazz) {
        this.clazz = clazz;
    }

    @Override
    public void setUrl(URL url) {
        this.url = url;
    }

    @Override
    public void setMethodAccess(MethodAccess methodAccess) {
        this.methodAccess = methodAccess;
    }

    public void setShare(boolean share) {
        this.share = share;
    }
}


<|start_filename|>demo/democlient/src/main/java/com/colobu/rpcx/config/ConsumerConfig.java<|end_filename|>
package com.colobu.rpcx.config;


import com.colobu.rpcx.fail.FailType;
import com.colobu.rpcx.selector.SelectMode;
import com.colobu.rpcx.service.IArith;
import com.colobu.rpcx.service.ITestService;
import com.colobu.rpcx.spring.RpcxConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.colobu.rpcx.rpc.impl.ConsumerConfig.ConsumerConfigBuilder;

/**
 * @author <EMAIL>
 */
@Configuration
public class ConsumerConfig {

    @Autowired
    private RpcxConsumer rpcxConsumer;

    @Bean
    public IArith arith() {
        return rpcxConsumer.wrap(IArith.class);
    }

    @Bean
    public ITestService testService() {
        return rpcxConsumer.wrap(ITestService.class, new ConsumerConfigBuilder()
//                .setGroup("test")
//                .setToken("<PASSWORD>")
                .setSendType(Constants.SYNC_KEY)
                .setTimeout(600000)
                .setFailType(FailType.FailTry)
                .setSelectMode(SelectMode.WeightedRoundRobin)
                .setRetryNum(3));
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/Exporter.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.common.Config;
import com.colobu.rpcx.common.NetUtils;
import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.FilterWrapper;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.annotation.Provider;
import com.esotericsoftware.reflectasm.MethodAccess;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author <EMAIL>
 */
public class Exporter {

    private static final Logger logger = LoggerFactory.getLogger(Exporter.class);

    public static final Map<String, Invoker<Object>> invokerMap = new HashMap<>(20);

    private final Function<Class, Object> getBeanFunc;

    private final String addr;

    public Exporter(Function<Class, Object> getBeanFunc, String addr) {
        this.getBeanFunc = getBeanFunc;
        this.addr = addr;
    }


    /**
     * 导出所有对外服务(包括 echo 和 泛化调用)
     *
     * @param providerPackage
     * @return
     */
    public Set<String> export(String providerPackage) {
        logger.info("export");
        String[] ss = addr.split(":");
        final String host = ss[0];
        final int port = Integer.valueOf(ss[1]);
        Set<Class<?>> set = new ProviderFinder(providerPackage).find();
        return set.stream().map(it -> {
            String name = it.getName();

            Method[] methods = it.getDeclaredMethods();
            Arrays.stream(methods).forEach(m -> {
                String className = it.getName();
                String method = m.getName();
                //reflectasm 生成的方法需要过滤掉
                if (!method.startsWith("access$")) {
                    String[] parameterTypeNames = ClassUtils.getMethodParameterNames(m);
                    String key = ClassUtils.getMethodKey(className, method, parameterTypeNames);
                    logger.info("export key:{}", key);
                    URL url = new URL("rpcx", host, port, key);
                    //暴露出invoker,使得远程可以调用
                    initProviderInvoker(key, className, method, parameterTypeNames, url);
                }
            });
            //echo invoker
            initEchoInvoker();
            //generic invoker
            initGenericInvoker();

            //这里的url是注册到zk
            URL url = new URL("rpcx", host, port);
            url.setPath(name);
            //类名称
            Provider provider = it.getAnnotation(Provider.class);
            //抽取出权重
            String weight = provider.weight();
            String weightValue = Config.ins().get(weight);
            if (null == weightValue) {
                weightValue = "1";
            }
            String group = "";
            if (null != provider.group()) {
                group = provider.group();
            }
            //导出的时候需要确定权重
            url = url.addParameter("weight", weightValue);
            url = url.addParameter("group", group);
            return url.toFullString();
        }).collect(Collectors.toSet());
    }


    /**
     * 初始化providerInvoker
     */
    private Invoker<Object> initProviderInvoker(String key, String className, String methodName, String[] parameterTypeNames, URL url) {
        Invoker<Object> wrapperInvoker;
        Invoker<Object> invoker = new RpcProviderInvoker<>(getBeanFunc);
        Class clazz = ClassUtils.getClassByName(className);
        invoker.setInterface(clazz);

        invoker.setUrl(url);

        //类级别的注解
        Provider typeProvider = invoker.getInterface().getAnnotation(Provider.class);
        setTypeParameters(typeProvider, url.getParameters());
        Set<String> excludeFilters = Arrays.stream(typeProvider.excludeFilters()).collect(Collectors.toSet());

        //兼容性更好些
        Method method = ClassUtils.getMethod(className, methodName, parameterTypeNames);
        invoker.setMethod(method);

        //reflectasm更快些(实际上是把反射生成了直接调用的代码)
        MethodAccess methodAccess = MethodAccess.get(ClassUtils.getClassByName(className));
        invoker.setMethodAccess(methodAccess);

        //方法级别的注解
        Provider methodProvider = method.getAnnotation(Provider.class);

        //方法级别的会覆盖type级别的
        if (null != methodProvider) {
            setMethodParameters(methodProvider, url.getParameters());
        }

        wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.PROVIDER, excludeFilters);

        invokerMap.putIfAbsent(key, wrapperInvoker);
        return wrapperInvoker;
    }

    /**
     * echo
     */
    private void initEchoInvoker() {
        Invoker<Object> invoker = new RpcProviderInvoker<>(getBeanFunc);
        URL url = new URL("rpcx", "", 0);
        url.setHost(NetUtils.getLocalHost());
        url.setPort(0);
        invoker.setUrl(url);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.PROVIDER, Sets.newHashSet());
        invokerMap.putIfAbsent(Constants.$ECHO, wrapperInvoker);
    }

    /**
     * 泛化调用
     */
    private void initGenericInvoker() {
        Invoker<Object> invoker = new RpcProviderInvoker<>(getBeanFunc);
        URL url = new URL("rpcx", "", 0);
        url.setHost(NetUtils.getLocalHost());
        url.setPort(0);
        invoker.setUrl(url);
        ((RpcProviderInvoker<Object>) invoker).setShare(true);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.PROVIDER, Sets.newHashSet());
        invokerMap.putIfAbsent(Constants.$INVOKE, wrapperInvoker);
    }

    private void setTypeParameters(Provider provider, Map<String, String> parameters) {
        parameters.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE);
        parameters.put(Constants.TOKEN_KEY, provider.token());
        parameters.put(Constants.TIMEOUT_KEY, String.valueOf(provider.timeout()));
        parameters.put(Constants.CACHE_KEY, String.valueOf(provider.cache()));
        parameters.put(Constants.TPS_LIMIT_RATE_KEY, String.valueOf(provider.tps()));
        parameters.put(Constants.MONITOR_KEY, String.valueOf(provider.monitor()));
        parameters.put(Constants.GROUP_KEY, String.valueOf(provider.group()));
        parameters.put(Constants.VERSION_KEY, String.valueOf(provider.version()));
    }

    private void setMethodParameters(Provider provider, Map<String, String> parameters) {
        //使用了那些注解,实际的含义就是启用了那些filter
        if (StringUtils.isNotEmpty(provider.token())) {
            parameters.put(Constants.TOKEN_KEY, provider.token());
        }
        if (-1 != provider.timeout()) {
            parameters.put(Constants.TIMEOUT_KEY, String.valueOf(provider.timeout()));
        }
        if (false != provider.cache()) {
            parameters.put(Constants.CACHE_KEY, String.valueOf(provider.cache()));
        }
        if (-1 != provider.tps()) {
            parameters.put(Constants.TPS_LIMIT_RATE_KEY, String.valueOf(provider.tps()));
        }
        if (false != provider.monitor()) {
            parameters.put(Constants.MONITOR_KEY, String.valueOf(provider.monitor()));
        }
        if (StringUtils.isNotEmpty(provider.group())) {
            parameters.put(Constants.GROUP_KEY, String.valueOf(provider.group()));
        }
        if (StringUtils.isNotEmpty(provider.version())) {
            parameters.put(Constants.VERSION_KEY, String.valueOf(provider.version()));
        }
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/selector/WeightedRountRobin.java<|end_filename|>
package com.colobu.rpcx.selector;

import com.colobu.rpcx.rpc.URL;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author <EMAIL>
 * <p>
 * https://github.com/phusion/nginx/commit/27e94984486058d73157038f7950a0a36ecc6e35
 */
public class WeightedRountRobin {

    private List<Weighted> weighteds;

    public WeightedRountRobin(Weighted[] weighteds) {
        this.weighteds = Arrays.asList(weighteds);
    }


    private boolean contains(URL url) {
        String weight = url.getParameter("weight");
        String address = url.getAddress();

        return weighteds.stream().anyMatch((it) -> {
            if (it.getServer().equals(address) && it.getWeight() == Integer.parseInt(weight)) {
                return true;
            }
            return false;
        });

    }


    private boolean contains(List<String> serverList, Weighted weighted) {
        return serverList.stream().anyMatch(it -> {
            URL url = URL.valueOf(it);
            if (url.getAddress().equals(weighted.server) && url.getParameter("weight").equals(String.valueOf(weighted.weight))) {
                return true;
            }
            return false;
        });
    }


    /**
     * 有可能会有服务器下线或上线
     *
     * @param serviceList
     */
    public void updateWeighteds(List<String> serviceList) {

        this.weighteds = weighteds.stream().filter(it -> {
            return contains(serviceList, it);
        }).collect(Collectors.toList());

        serviceList.stream().forEach(it -> {
            URL url = URL.valueOf(it);
            if (!contains(url)) {
                weighteds.add(new Weighted(url.getAddress(), Integer.parseInt(url.getParameter("weight"))));
            }
        });

    }


    public WeightedRountRobin(List<String> serviceList) {
        Weighted[] weighteds = serviceList.stream().map(it -> {
            URL url = URL.valueOf(it);
            Weighted weighted = new Weighted(url.getAddress(), Integer.parseInt(url.getParameter("weight")));
            return weighted;
        }).toArray(Weighted[]::new);
        this.weighteds = Arrays.asList(weighteds);
    }

    public Weighted nextWeighted() {
        Weighted best = null;

        int total = 0;

        for (int i = 0; i < weighteds.size(); i++) {
            Weighted w = weighteds.get(i);

            if (w == null) {
                continue;
            }
            //if w is down, continue

            w.currentWeight += w.effectiveWeight;
            total += w.effectiveWeight;
            if (w.effectiveWeight < w.weight) {
                w.effectiveWeight++;
            }

            if (best == null || w.currentWeight > best.currentWeight) {
                best = w;
            }

        }

        if (best == null) {
            return null;
        }

        best.currentWeight -= total;
        return best;
    }
}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/bootstrap/Bootstrap.java<|end_filename|>
package com.colobu.rpcx.bootstrap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.web.bind.annotation.RestController;


/**
 * @author <EMAIL>
 */
@EnableAutoConfiguration
@ComponentScan(basePackages = {"com.colobu.rpcx"})
@RestController
public class Bootstrap {

    private static final Logger logger = LoggerFactory.getLogger(Bootstrap.class);

    public static void main(String... args) {
        try {
            SpringApplication.run(Bootstrap.class, args);
        } catch (Throwable ex) {
            logger.error(ex.getMessage(), ex);
            System.exit(0);
        }
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/RpcConsumerInvoker.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.netty.IClient;
import com.colobu.rpcx.protocol.*;
import com.colobu.rpcx.rpc.*;
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author <EMAIL>
 */
public class RpcConsumerInvoker<T> implements Invoker<T> {

    private static final Logger logger = LoggerFactory.getLogger(RpcConsumerInvoker.class);

    private static final AtomicInteger seq = new AtomicInteger();

    private IClient client;

    private URL url;

    public RpcConsumerInvoker(IClient client) {
        this.client = client;
    }

    @Override
    public Class<T> getInterface() {
        return null;
    }

    @Override
    public Result invoke(RpcInvocation invocation) throws RpcException {
        String className = invocation.getClassName();
        String method = invocation.getMethodName();
        RpcResult result = new RpcResult();

        Message req = new Message(className, method);
        req.setVersion((byte) 0);
        req.setMessageType(MessageType.Request);
        req.setHeartbeat(false);
        req.setOneway(false);
        req.setCompressType(CompressType.None);
        req.setSerializeType(invocation.getSerializeType());

        invocation.setUrl(this.url);
        byte[] data = null;
        if (invocation.getLanguageCode().equals(LanguageCode.GO)) {
            req.metadata.put(Constants.LANGUAGE, LanguageCode.GO.name());
            data = invocation.getPayload();
        } else if (invocation.getLanguageCode().equals(LanguageCode.JAVA)) {
            req.metadata.put(Constants.LANGUAGE, LanguageCode.JAVA.name());
            data = HessianUtils.write(invocation);
        }
        req.payload = data;

        try {
            req.setSeq(seq.incrementAndGet());
            Message res = client.call(req, invocation.getTimeOut(), invocation.getSendType());

            result.getAttachments().put(Constants.TRACE_ID,res.getMetadata().get(Constants.TRACE_ID));
            RpcContext.getContext().getAttachments().put(Constants.TRACE_ID,res.getMetadata().get(Constants.TRACE_ID));
            result.getAttachments().put(Constants.SPAN_ID,res.getMetadata().get(Constants.SPAN_ID));
            RpcContext.getContext().getAttachments().put(Constants.SPAN_ID,res.getMetadata().get(Constants.SPAN_ID));

            if (res.metadata.containsKey(Constants.RPCX_ERROR_CODE)) {
                String code = (res.metadata.get(Constants.RPCX_ERROR_CODE));
                String message = res.metadata.get(Constants.RPCX_ERROR_MESSAGE);
                if (message == null) {
                    message = "";
                }
                logger.warn("client call error:{}:{}", code, message);
                RpcException error = new RpcException(message, code);
                result.setThrowable(error);
            } else {
                byte[] d = res.payload;
                if (d.length > 0) {
                    if (invocation.getLanguageCode().equals(LanguageCode.JAVA)) {
                        Object r = HessianUtils.read(d);
                        result.setValue(r);
                    }
                    if (invocation.getLanguageCode().equals(LanguageCode.GO)) {
                        result.setValue(d);
                    }
                }

            }
        } catch (Throwable e) {
            result.setThrowable(e);
            logger.info("client call error:{} ", e.getMessage());
        }

        logger.debug("class:{} method:{} result:{} finish", className, method, new Gson().toJson(result));
        return result;
    }

    @Override
    public void setMethod(Method method) {

    }

    @Override
    public Method getMethod() {
        return null;
    }

    @Override
    public void setInterface(Class clazz) {

    }

    @Override
    public URL getUrl() {
        return url;
    }

    @Override
    public void setUrl(URL url) {
        this.url = url;
    }

    @Override
    public boolean isAvailable() {
        return true;
    }

    @Override
    public void destroy() {

    }

    @Override
    public IServiceDiscovery serviceDiscovery() {
        return this.client.getServiceDiscovery();
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/tps/StatItem.java<|end_filename|>
package com.colobu.rpcx.tps;


import java.util.concurrent.atomic.AtomicInteger;

class StatItem {

    private String name;

    private long lastResetTime;

    private long interval;

    private AtomicInteger token;

    private int rate;

    StatItem(String name, int rate, long interval) {
        this.name = name;
        this.rate = rate;
        this.interval = interval;
        this.lastResetTime = System.currentTimeMillis();
        this.token = new AtomicInteger(rate);
    }

    public boolean isAllowable() {
        long now = System.currentTimeMillis();
        if (now > lastResetTime + interval) {//如果超过了限流时间,则数量和时间和rate都重置
            token.set(rate);
            lastResetTime = now;
        }

        int value = token.get();
        boolean flag = false;
        while (value > 0 && !flag) {//如果小于0了则直接返回false
            flag = token.compareAndSet(value, value - 1);
            value = token.get();
        }

        return flag;
    }

    long getLastResetTime() {
        return lastResetTime;
    }
    
    int getToken() {
        return token.get();
    }
    
    @Override
    public String toString() {
        return new StringBuilder(32).append("StatItem ")
            .append("[name=").append(name).append(", ")
            .append("rate = ").append(rate).append(", ")
            .append("interval = ").append(interval).append("]")
            .toString();
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/DefaultCommandHeader.java<|end_filename|>
package com.colobu.rpcx.protocol;


/**
 * Created by <EMAIL>.
 */
public class DefaultCommandHeader extends CommandCustomHeader {
}


<|start_filename|>demo/demoserver2/src/main/java/com/colobu/rpcx/service/Test2Service.java<|end_filename|>
package com.colobu.rpcx.service;

import com.colobu.rpcx.rpc.annotation.Provider;
import com.colobu.rpcx.service.ITest2Service;
import org.springframework.stereotype.Service;

/**
 * @author <EMAIL>
 */
@Service
@Provider(name = "Test2Service")
public class Test2Service implements ITest2Service {

    public int num() {
        return 100;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyEventType.java<|end_filename|>
package com.colobu.rpcx.netty;

/**
 * @author <EMAIL>
 */
public enum NettyEventType {
    /**
     * 连接
     */
    CONNECT,

    /**
     * 关闭
     */
    CLOSE,

    /**
     * 空闲
     */
    IDLE,

    /**
     * 异常
     */
    EXCEPTION
}


<|start_filename|>demo/democlient/src/main/java/com/colobu/rpcx/handler/TestController.java<|end_filename|>
package com.colobu.rpcx.handler;

import com.colobu.rpcx.netty.ResponseFuture;
import com.colobu.rpcx.rpc.RpcContext;
import com.colobu.rpcx.service.ITestService;
import com.colobu.rpcx.spring.RpcxConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;

/**
 * @author <EMAIL>
 */
@RestController
public class TestController {

    @Autowired
    private ITestService testService;

    @Autowired
    private ApplicationContext context;


    @Autowired
    private RpcxConsumer consumer;


    @GetMapping("/hi")
    public String input(String word) {
        String s = testService.hi(word);
        return s;
    }


    @GetMapping("/sum")
    public String sum() {
        String s = String.valueOf(testService.sum(22, 33));
        return s;
    }

    @GetMapping("/sum2")
    public String sum2() {
        String s = String.valueOf(testService.sum2(22, 33));
        return s;
    }

    /**
     * 测试echo filter
     *
     * @return
     */
    @GetMapping("/echo")
    public String echo() {
        String s = consumer.echo("com.colobu.rpcx.service.TestService", "echo");
        return s;
    }

    /**
     * 泛化调用
     *
     * @return
     */
    @GetMapping("/generic")
    public String generic() {
        Object s = consumer.invoke("com.colobu.rpcx.service.TestService", "sum", new String[]{"int", "int"}, new String[]{"11", "22"});
        return s.toString();
    }

    /**
     * 异步模式(必须consumer 开启 异步模式)
     * ConsumerConfigBuilder.setSendType(Constants.ASYNC_KEY)
     * @return
     * @throws InterruptedException
     */
    @GetMapping("/async")
    public String async() throws InterruptedException {
        testService.async();
        ResponseFuture<String> f1 = RpcContext.getContext().getFuture();
        //需要自己手动清理
        RpcContext.removeContext();
        testService.async();
        ResponseFuture<String> f2 = RpcContext.getContext().getFuture();
        RpcContext.removeContext();
        String result = "";
        if (null != f1 && f2 != null) {
            String s1 = f1.get(2000);
            String s2 = f2.get(2000);
            result = s1 + s2;
        }

        return result;
    }

    /**
     * 测试热更新
     *
     * @return
     * @throws IOException
     */
    @GetMapping("/deploy")
    public String deploy() throws IOException {
        Object s = consumer.deploy("com.colobu.rpcx.service.TestService", "/Users/zhangzhiyong/IdeaProjects/rpcx-java/demo/demmoserver/target/classes/com/colobu/rpcx/service/TestService.class", "123abc");
        return s.toString();
    }


}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/ZkServiceDiscovery.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.common.Pair;
import com.colobu.rpcx.concurrent.ConcurrentHashSet;
import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.rpc.impl.ConsumerFinder;
import com.colobu.rpcx.utils.PathStatus;
import com.colobu.rpcx.utils.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * @author <EMAIL>
 */
public class ZkServiceDiscovery implements IServiceDiscovery {

    private static final Logger logger = LoggerFactory.getLogger(ZkServiceDiscovery.class);

    private ConcurrentHashMap<String, Set<Pair<String, String>>> map = new ConcurrentHashMap<>();

    private final String basePath;

    private ConcurrentHashSet<String> serviceNameSet = new ConcurrentHashSet<>();

    private LinkedBlockingQueue<PathStatus> queue = new LinkedBlockingQueue<>();

    private Set<String> findConsumer(String consumerPackage) {
        return new ConsumerFinder().find(consumerPackage);
    }

    private AtomicBoolean stop = new AtomicBoolean(false);

    /**
     * 自动查找依赖的服务
     *
     * @param basePath
     */
    public ZkServiceDiscovery(final String basePath, String consumerPackage) {
        this.basePath = basePath;

        //有多个业务提供方
        Arrays.stream(consumerPackage.split(";")).forEach(it -> {
            this.serviceNameSet.addAll(findConsumer(it));
        });

        this.serviceNameSet.stream().forEach(it -> this.map.put(it, new HashSet<>()));
        zkServiceDiscovery(basePath);
    }


    private void zkServiceDiscovery(String basePath) {
        this.serviceNameSet.stream().forEach(it -> {
            Set<Pair<String, String>> set = ZkClient.ins().get(basePath, it).stream().map(it2 -> {
                String addr = "";
                try {
                    if (it2.getObject1().contains("@")) {
                        addr = it2.getObject1().split("@")[1];
                    } else {
                        addr = it2.getObject1();
                    }
                } catch (Exception ex) {

                }
                return Pair.of(addr, it2.getObject2());
            }).collect(Collectors.toSet());
            this.map.put(it, set);
        });

        //监控zk
        watch();
    }

    @Override
    public void addServices(final String... serviceNames) {
        Arrays.stream(serviceNames).forEach(it -> this.serviceNameSet.add(it));
        Arrays.stream(serviceNames).forEach(it -> {
            Set<Pair<String, String>> set = ZkClient.ins().get(basePath, it).stream().map(it2 -> {
                String addr = "";
                try {
                    if (it2.getObject1().contains("@")) {
                        addr = it2.getObject1().split("@")[1];
                    } else {
                        addr = it2.getObject1();
                    }
                } catch (Exception ex) {

                }
                return Pair.of(addr, it2.getObject2());
            }).collect(Collectors.toSet());
            this.map.putIfAbsent(it, set);
            try {
                ZkClient.ins().watch(queue, basePath + it);
            } catch (Exception e) {
                logger.error("service name foreach {} error:{}", it, e.getMessage());
            }
        });
    }

    /**
     * 根据提供的服务名字
     */
    public ZkServiceDiscovery(final String basePath, LanguageCode languageCode, final String... serviceNames) {
        logger.info("ZkServiceDiscovery languageCode:{}", languageCode);
        this.basePath = basePath;
        Arrays.stream(serviceNames).forEach(it -> this.serviceNameSet.add(it));
        zkServiceDiscovery(basePath);
    }


    @Override
    public List<String> getServices(String serviceName) {
        return this.map.get(serviceName).stream().map(it -> it.getObject1() + "?" + it.getObject2()).collect(Collectors.toList());
    }


    @Override
    public void watch() {
        new Thread(() -> {
            while (true) {
                try {
                    PathStatus ps = queue.take();
                    if (ps.isStop()) {
                        break;
                    }
                    logger.info("=========>zk discovery ps:{}", ps);
                    String service = ps.getPath().replace(this.basePath, "");

                    if (this.map.containsKey(service)) {
                        if ("CHILD_ADDED".equals(ps.getType())) {
                            this.map.compute(service, (k, v) -> {
                                v.add(ps.getValue());
                                return v;
                            });
                        } else if ("CHILD_REMOVED".equals(ps.getType())) {
                            this.map.compute(service, (k, v) -> {
                                v.remove(ps.getValue());
                                return v;
                            });
                        } else if ("CHILD_UPDATED".equals(ps.getType())) {
                            this.map.compute(service, (k, v) -> {
                                v = v.stream().map(it -> {
                                    if (it.getObject1().equals(ps.getValue().getObject1())) {
                                        it.setObject2(ps.getValue().getObject2());
                                        return it;
                                    }
                                    return it;
                                }).collect(Collectors.toSet());
                                return v;
                            });
                        }

                    }

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        //根据serviceName 进行监控
        this.serviceNameSet.stream().forEach(it -> {
            try {
                ZkClient.ins().watch(queue, basePath + it);
            } catch (Exception e) {
                logger.error("service name foreach {} error:{}", it, e.getMessage());
            }
        });
    }

    @Override
    public void close() {
        this.queue.offer(new PathStatus(true));
        ZkClient.ins().close();
        stop.compareAndSet(false, true);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/config/RpcxConfig.java<|end_filename|>
package com.colobu.rpcx.config;

public class RpcxConfig {
    /**
     * consumer的包路径
     */
    private String consumerPackage;
    /**
     * provider 的包路径
     */
    private String providerPackage;
    /**
     * 用户自定义filter 的包路径
     */
    private String filterPackage;
    /**
     * rpcx 连接 zk的地址 host:port
     */
    private String zkAddr;
    /**
     * 热更新的token
     */
    private String hotDeployToken;
    /**
     * 服务的端口号
     */
    private int serverPort;

    /**
     * 业务处理的线程数量
     */
    private int bizProcessorNum;

    /**
     * 服务器read write的线程数量
     */
    private int workerProcessorNum;

    private int sndBufSize;

    private int rcvBufSize;


    public String getConsumerPackage() {
        return consumerPackage;
    }

    public void setConsumerPackage(String consumerPackage) {
        this.consumerPackage = consumerPackage;
    }

    public String getProviderPackage() {
        return providerPackage;
    }

    public void setProviderPackage(String providerPackage) {
        this.providerPackage = providerPackage;
    }

    public String getFilterPackage() {
        return filterPackage;
    }

    public void setFilterPackage(String filterPackage) {
        this.filterPackage = filterPackage;
    }

    public String getZkAddr() {
        return zkAddr;
    }

    public void setZkAddr(String zkAddr) {
        this.zkAddr = zkAddr;
    }

    public String getHotDeployToken() {
        return hotDeployToken;
    }

    public void setHotDeployToken(String hotDeployToken) {
        this.hotDeployToken = hotDeployToken;
    }

    public int getServerPort() {
        return serverPort;
    }

    public void setServerPort(int serverPort) {
        this.serverPort = serverPort;
    }


    public int getBizProcessorNum() {
        return bizProcessorNum;
    }

    public void setBizProcessorNum(int bizProcessorNum) {
        this.bizProcessorNum = bizProcessorNum;
    }

    public int getWorkerProcessorNum() {
        return workerProcessorNum;
    }

    public void setWorkerProcessorNum(int workerProcessorNum) {
        this.workerProcessorNum = workerProcessorNum;
    }

    public int getSndBufSize() {
        return sndBufSize;
    }

    public void setSndBufSize(int sndBufSize) {
        this.sndBufSize = sndBufSize;
    }

    public int getRcvBufSize() {
        return rcvBufSize;
    }

    public void setRcvBufSize(int rcvBufSize) {
        this.rcvBufSize = rcvBufSize;
    }
}


<|start_filename|>server/src/main/java/com/colobu/rpcx/processor/RpcProcessor.java<|end_filename|>
package com.colobu.rpcx.processor;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.netty.NettyRequestProcessor;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.protocol.RemotingSysResponseCode;
import com.colobu.rpcx.rpc.HessianUtils;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.impl.Exporter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;

/**
 * @author <EMAIL>
 * 处理远程rpc调用
 */
public class RpcProcessor implements NettyRequestProcessor {

    private static final Logger logger = LoggerFactory.getLogger(RpcProcessor.class);

    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws UnsupportedEncodingException {
        request.getMessage().decode(request.getData());

        String language = request.getMessage().metadata.get(Constants.LANGUAGE);
        String className = request.getMessage().servicePath;
        String methodName = request.getMessage().serviceMethod;
        RpcInvocation invocation = null;
        //golang调用是没有language的
        if (null == language) {
            //golang 是没有invocation的
            invocation = new RpcInvocation();
            invocation.setClassName(className);
            invocation.setMethodName(methodName);
            invocation.setLanguageCode(LanguageCode.GO);
            invocation.setParameterTypeNames(new String[]{"byte[]"});
            String path = ClassUtils.getMethodKey(className, methodName, new String[]{"byte[]"});
            invocation.url = new URL("rpcx", "", 0, path);
            //golang 参数是byte[]
            invocation.setParameterTypeNames(new String[]{"byte[]"});
            //参数就是payload数据
            invocation.setArguments(new Object[]{request.getMessage().payload});
        } else if (LanguageCode.JAVA.name().equals(language)) {
            invocation = (RpcInvocation) HessianUtils.read(request.getMessage().payload);
        }

        Invoker<Object> wrapperInvoker = null;

        //request -> response
        RemotingCommand res = request.requestToResponse();

        String key = "";


        if (methodName.equals(Constants.$ECHO)) {
            //echo
            key = Constants.$ECHO;
        } else if (methodName.equals(Constants.$INVOKE)) {
            //泛化调用
            key = Constants.$INVOKE;
        } else {
            key = ClassUtils.getMethodKey(className, methodName, invocation.getParameterTypeNames());
        }

        wrapperInvoker = Exporter.invokerMap.get(key);

        if (null == wrapperInvoker) {
            logger.warn("get invoker is null key:{}", key);
            res.setErrorMessage(RemotingSysResponseCode.SYSTEM_ERROR, "get invoker is null  key:" + key);
            return res;
        }

        Result rpcResult = wrapperInvoker.invoke(invocation);
        res.getMessage().metadata.put(Constants.TRACE_ID, rpcResult.getAttachment(Constants.TRACE_ID, ""));
        res.getMessage().metadata.put(Constants.SPAN_ID, rpcResult.getAttachment(Constants.SPAN_ID, ""));

        if (invocation.languageCode.equals(LanguageCode.JAVA)) {
            res.getMessage().payload = HessianUtils.write(rpcResult.getValue());
        } else {
            res.getMessage().payload = (byte[]) rpcResult.getValue();
        }
        if (rpcResult.hasException()) {
            logger.error(rpcResult.getException().getMessage(), rpcResult.getException());
            res.setErrorMessage(RemotingSysResponseCode.SYSTEM_ERROR, rpcResult.getException().getMessage());
        }
        return res;
    }


    @Override
    public boolean rejectRequest() {
        return false;
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/Filter.java<|end_filename|>
package com.colobu.rpcx.filter;

import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.impl.RpcInvocation;

/**
 * @author <EMAIL>
 */
public interface Filter {

    Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException;

    /**
     * filter name
     * @return
     */
    default String name() {
        return this.getClass().getSimpleName();
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/GenericFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.ReflectUtils;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.Exporter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.gson.Gson;

import java.lang.reflect.Method;
import java.util.stream.IntStream;


/**
 * @author <EMAIL>
 * 泛化调用
 */
@RpcFilter(group = {Constants.PROVIDER}, order = -2001)
public class GenericFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation inv) throws RpcException {
        if (inv.getMethodName().equals(Constants.$INVOKE)) {
            String[] types = inv.getParameterTypeNames();
            final String[] args = (String[]) inv.getArguments();

            Gson gson = new Gson();
            int len = types.length;

            Class[] classTypes = new Class[types.length];

            Object[] params = IntStream.range(0, len).mapToObj(i -> {
                String type = types[i];
                Class<?> clazz = ReflectUtils.forName(type);
                classTypes[i] = clazz;
                int j = i + 1;
                String arg = args[j];
                Object obj = gson.fromJson(arg, clazz);
                return obj;
            }).toArray(Object[]::new);

            //从新组装Invocation
            inv.setMethodName(args[0]);
            inv.setParameterTypes(classTypes);
            inv.setArguments(params);


            Method method = ClassUtils.getMethod(inv.getClassName(), inv.getMethodName(), inv.getParameterTypeNames());
            invoker.setMethod(method);

            Result res = invoker.invoke(inv);

            if (!res.hasException()) {
                Object value = res.getValue();
                res.setValue(gson.toJson(value));
                return res;
            }

        }

        return invoker.invoke(inv);
    }

}

<|start_filename|>common/src/test/java/com/colobu/rpcx/common/TestPoolFactory.java<|end_filename|>
package com.colobu.rpcx.common;

import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.RemotingCommand;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;

public class TestPoolFactory extends BasePooledObjectFactory<RemotingCommand> {

    @Override
    public RemotingCommand create() {
        return RemotingCommand.createRequestCommand(new Message());
    }

    @Override
    public PooledObject<RemotingCommand> wrap(RemotingCommand remotingCommand) {
        return new DefaultPooledObject<>(remotingCommand);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/monitor/MonitorFactory.java<|end_filename|>
package com.colobu.rpcx.monitor;

import com.colobu.rpcx.monitor.impl.RpcxMonitor;
import com.colobu.rpcx.rpc.URL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;

/**
 * @author <EMAIL>
 */
public class MonitorFactory {

    private static final Logger logger = LoggerFactory.getLogger(MonitorFactory.class);

    /**
     * Create monitor.
     */
    public Monitor getMonitor(URL url) {
        return new RpcxMonitor(new MonitorService() {
            @Override
            public void collect(URL statistics) {
                logger.info(statistics.toFullString());
            }

            @Override
            public List<URL> lookup(URL query) {
                return null;
            }
        });
    }

}

<|start_filename|>client/src/main/java/com/colobu/rpcx/client/impl/RandomSelector.java<|end_filename|>
package com.colobu.rpcx.client.impl;

import com.colobu.rpcx.client.Selector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Random;

/**
 * Created by <EMAIL>.
 */
public class RandomSelector implements Selector {

    private static final Logger logger = LoggerFactory.getLogger(RandomSelector.class);

    @Override
    public String select(String servicePath, String serviceMethod, List<String> list) {
        if (list.size() > 0) {
            Random random = new Random();
            int i = random.nextInt(list.size());
            String addr = list.get(i);
            logger.info("random select addr:{}", addr);
            return addr;
        }
        return null;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/SelectorFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.*;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.selector.SelectMode;
import com.colobu.rpcx.selector.Weighted;
import com.colobu.rpcx.selector.WeightedRountRobin;

import java.security.SecureRandom;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


/**
 * @author <EMAIL>
 */
@RpcFilter(group = {Constants.CONSUMER})
public class SelectorFilter implements Filter {

    private static SecureRandom secureRandom = new SecureRandom();

    private static ConcurrentHashMap<String, Integer> selectIndexMap = new ConcurrentHashMap<>();

    /**
     * 基于权重
     */
    private static ConcurrentHashMap<String, WeightedRountRobin> weightRoundRobinMap = new ConcurrentHashMap<>();

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        //目前使用类名当服务名称
        String serviceName = invocation.getClassName();

        //获取服务列表(ip:port?param)
        List<String> serviceList = invoker.serviceDiscovery().getServices(serviceName);

        serviceList = serviceList.stream().filter(it->{
            URL url = URL.valueOf(it);
            if (!it.contains("state=inactive") && invocation.getGroup().equals(url.getParameter(Constants.GROUP_KEY,""))) {
                return true;
            }
            return false;
        }).collect(Collectors.toList());


        if (serviceList.size() <= 0) {
            throw new RpcException("service list size <=0 ", "-3");
        }

        SelectMode selectMode = invocation.getSelectMode();
        switch (selectMode) {
            case RandomSelect: {
                String service = randomSelect(serviceList);
                RpcContext.getContext().setServiceAddr(service);
                break;
            }
            case RoundRobin: {
                String service = roundRobin(serviceList, serviceName);
                RpcContext.getContext().setServiceAddr(service);
                break;
            }
            case WeightedRoundRobin: {
                String service = weightedRountRobin(serviceList, serviceName);
                RpcContext.getContext().setServiceAddr(service);
                break;
            }
            case SelectByUser: {
                break;
            }
        }


        return invoker.invoke(invocation);
    }


    private String roundRobin(List<String> serviceList, String serviceName) {
        Integer index = selectIndexMap.get(serviceName);
        if (null == index) {
            selectIndexMap.putIfAbsent(serviceName, 0);
            index = selectIndexMap.get(serviceName);
        }

        int i = index % serviceList.size();
        String serviceAddr = serviceList.get(i);
        selectIndexMap.compute(serviceName, (k, v) -> {
            v = v + 1;
            return v;
        });
        return URL.valueOf(serviceAddr).getAddress();
    }


    private String weightedRountRobin(List<String> serviceList, String serviceName) {
        WeightedRountRobin weightedRountRobin = weightRoundRobinMap.get(serviceName);
        if (null == weightedRountRobin) {
            weightRoundRobinMap.putIfAbsent(serviceName, new WeightedRountRobin(serviceList));
            weightedRountRobin = weightRoundRobinMap.get(serviceName);
        }
        //需要更新权重列表,有可能服务器上下线
        weightedRountRobin.updateWeighteds(serviceList);

        Weighted wei = weightedRountRobin.nextWeighted();
        if (null == wei) {
            return null;
        }
        return URL.valueOf(wei.getServer()).getAddress();
    }


    private String randomSelect(List<String> serviceList) {
        int index = secureRandom.nextInt(serviceList.size());
        return URL.valueOf(serviceList.get(index)).getAddress();
    }
}


<|start_filename|>server/src/main/java/com/colobu/rpcx/handler/RpcxProcessHandler.java<|end_filename|>
package com.colobu.rpcx.handler;

import com.colobu.rpcx.netty.NettyConnetManageHandler;
import com.colobu.rpcx.netty.NettyDecoder;
import com.colobu.rpcx.netty.NettyEncoder;
import com.colobu.rpcx.netty.NettyServerHandler;
import com.colobu.rpcx.server.NettyServer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.List;

/**
 * @author <EMAIL>
 */
public class RpcxProcessHandler extends ByteToMessageDecoder {

    private int serverChannelMaxIdleTimeSeconds;

    private NettyServer nettyServer;

    public RpcxProcessHandler(int serverChannelMaxIdleTimeSeconds, NettyServer nettyServer) {
        this.serverChannelMaxIdleTimeSeconds = serverChannelMaxIdleTimeSeconds;
        this.nettyServer = nettyServer;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {

        if (in.readableBytes() < 1) {
            return;
        }

        final int magic = in.getByte(in.readerIndex());
        ChannelPipeline p = ctx.pipeline();
        //处理http
        if (isHttp(magic)) {
            p.addLast(new HttpServerCodec());
            p.addLast(new HttpObjectAggregator(1048576));
            p.addLast(new RpcxHttpHandler(nettyServer));
            p.remove(this);
        } else {//处理二进制
            p.addLast(new NettyEncoder());
            p.addLast(new NettyDecoder());
            p.addLast(new IdleStateHandler(0, 0, serverChannelMaxIdleTimeSeconds));
            p.addLast(new NettyConnetManageHandler(nettyServer));
            p.addLast(new NettyServerHandler(nettyServer));
            p.remove(this);
        }

    }


    // G for GET, and P for POST
    private static boolean isHttp(int magic) {
        return magic == 'G' || magic == 'P';
    }
}


<|start_filename|>demo/democlient/src/main/java/com/demo/client/filter/DemoClientFilter.java<|end_filename|>
package com.demo.client.filter;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author <EMAIL>
 */
@RpcFilter(group = {Constants.CONSUMER})
public class DemoClientFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(DemoClientFilter.class);

    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        logger.info("DemoClientFilter invoke");
        return invoker.invoke(invocation);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/MessageType.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum MessageType {
    /**
     * 请求
     */
    Request(0),

    /**
     * 应答
     */
    Response(1);


    private final int value;

    MessageType(int value) {
        this.value = value;
    }

    private static MessageType[] values = MessageType.values();
    public static MessageType getValue(int i) {
        return values[i];
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/config/Constants.java<|end_filename|>
package com.colobu.rpcx.config;

import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

/**
 * @author <EMAIL>
 */
public class Constants {

    public static final String VERSION_KEY = "version";

    public static final String GROUP_KEY = "group";

    public static final String BACKUP_KEY = "backup";

    public static final String DEFAULT_KEY_PREFIX = "default.";

    public static final Pattern COMMA_SPLIT_PATTERN = Pattern.compile("\\s*[,]+\\s*");

    public static final String INTERFACE_KEY = "interface";

    public static final String LOCALHOST_KEY = "localhost";

    public static final String ANYHOST_VALUE = "0.0.0.0";

    public static final String ANYHOST_KEY = "anyhost";

    public static final String ASYNC_KEY = "async";

    public static final String SYNC_KEY = "sync";

    public static final String ONE_WAY_KEY = "oneway";

    public static final String RETURN_KEY = "return";

    public static final String $ECHO = "$echo";

    public static final String $HOT_DEPLOY = "$hot_deploy";

    public static final String TOKEN_KEY = "token";

    public static final String TIMEOUT_KEY = "timeout";

    public static final String PATH_KEY = "path";

    public static final String CACHE_KEY = "cache";

    public static final String ACCESSLOG_KEY = "accesslog";

    public static final String FALSE = "false";

    public static final String TRUE = "true";

    public static final String COMMA_SEPARATOR = ",";

    public static final String TPS_LIMIT_RATE_KEY = "tps";

    public static final String TPS_LIMIT_INTERVAL_KEY = "tps.interval";

    public static final long DEFAULT_TPS_LIMIT_INTERVAL = TimeUnit.MINUTES.toMillis(1);

    public static final String MONITOR_KEY = "monitor";

    public static final String APPLICATION_KEY = "application";

    public static final String INPUT_KEY = "input";

    public static final String OUTPUT_KEY = "output";

    public static final String SIDE_KEY = "side";

    public static final String PROVIDER_SIDE = "provider";

    public static final String CONSUMER_SIDE = "consumer";

    public static final String COUNT_PROTOCOL = "count";

    public static final String PROVIDER = "provider";

    public static final String CONSUMER = "consumer";

    public static final String $INVOKE = "$invoke";

    public static final String SEND_TYPE = "sendType";

    public static final String FAIL_TYPE = "fail.type";

    public static final String LANGUAGE = "language";

    public static final String RPCX_ERROR_CODE = "__rpcx_error__";

    public static final String RPCX_ERROR_MESSAGE = "_rpcx_error_message";

    public static final String TRACE_ID = "trace_id";

    public static final String SPAN_ID = "span_id";

    public static final String X_RPCX_SERVICEPATH = "X-RPCX-ServicePath";

    public static final String X_RPCX_SERVICEMETHOD = "X-RPCX-ServiceMethod";

    public static final String X_RPCX_TRACEID = "X-RPCX-TraceId";


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/ConsumerContextFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.common.NetUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.rpc.*;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;


/**
 * @author <EMAIL>
 * 初始化consumer 上下文
 */
@RpcFilter(group = {Constants.CONSUMER}, order = -3000)
public class ConsumerContextFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        String path = "";

        if (invocation.getLanguageCode().equals(LanguageCode.JAVA)) {
            path = ClassUtils.getMethodKey(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
        }
        URL url = new URL("rpcx", NetUtils.getLocalAddress().getHostAddress(), 0, path);
        url = url.setServiceInterface(invocation.getClassName() + "." + invocation.getMethodName());
        invoker.setUrl(url);
        RpcContext.getContext()
                .setLocalAddress(NetUtils.getLocalHost(), 0)
                .setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort());
        invocation.setInvoker(invoker);
        try {
            return invoker.invoke(invocation);
        } finally {
            //同步的和oneway的直接清除掉context
            if (invocation.getSendType().equals(Constants.SYNC_KEY) || invocation.getSendType().equals(Constants.ONE_WAY_KEY)) {
                RpcContext.removeContext();
            } else {
                //async 需要手动自己清理(RpcContext.removeContext),因为future 还要从 contex他中拿
                RpcContext.getContext().clearAttachments();
                RpcContext.getContext().setServiceAddr("");
                RpcContext.getContext().setLocalAddress("", 0);
                RpcContext.getContext().setRemoteAddress("", 0);
            }
        }
    }

}

<|start_filename|>client/src/test/java/com/colobu/rpcx/client/RpcxHttpClientTest.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.client.http.RpcxHttpClient;
import com.colobu.rpcx.common.Pair;
import org.junit.Test;

public class RpcxHttpClientTest {


    @Test
    public void testExecute() {
        String res = RpcxHttpClient.execute("http://10.231.39.214:8033/", "com.colobu.rpcx.service.TestService", "sum", Pair.of("int", "11"), Pair.of("int", "22"));
        System.out.println(res);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/selector/Weighted.java<|end_filename|>
package com.colobu.rpcx.selector;

public class Weighted {

    public String server;
    public int weight;
    public int currentWeight;
    public int effectiveWeight;

    public Weighted(String server, int weight) {
        this.server = server;
        this.weight = weight;
    }

    public String getServer() {
        return server;
    }

    public void setServer(String server) {
        this.server = server;
    }

    public int getWeight() {
        return weight;
    }

    public void setWeight(int weight) {
        this.weight = weight;
    }

    public int getCurrentWeight() {
        return currentWeight;
    }

    public void setCurrentWeight(int currentWeight) {
        this.currentWeight = currentWeight;
    }

    public int getEffectiveWeight() {
        return effectiveWeight;
    }

    public void setEffectiveWeight(int effectiveWeight) {
        this.effectiveWeight = effectiveWeight;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyRequestProcessor.java<|end_filename|>

package com.colobu.rpcx.netty;

import com.colobu.rpcx.protocol.RemotingCommand;
import io.netty.channel.ChannelHandlerContext;

/**
 * @author <EMAIL>
 */
public interface NettyRequestProcessor {
    RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
            throws Exception;
    boolean rejectRequest();
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/InvokeCallback.java<|end_filename|>
package com.colobu.rpcx.common;


import com.colobu.rpcx.netty.ResponseFuture;

/**
 * @author <EMAIL>
 */
public interface InvokeCallback {
    void operationComplete(final ResponseFuture responseFuture);
}


<|start_filename|>container/src/main/java/com/colobu/rpcx/spring/RpcxConsumer.java<|end_filename|>
package com.colobu.rpcx.spring;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.FilterWrapper;
import com.colobu.rpcx.netty.IClient;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.ReflectUtils;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.impl.ConsumerConfig;
import com.colobu.rpcx.rpc.impl.RpcConsumerInvoker;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

/**
 * @author <EMAIL>
 */
public class RpcxConsumer {

    private static final Logger logger = LoggerFactory.getLogger(RpcxConsumer.class);


    private final IClient client;

    public RpcxConsumer(IClient client) {
        this.client = client;
    }

    /**
     * 全部使用默认配置
     *
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> T wrap(Class clazz) {
        return (T) new ConsumerConfig(client).refer(clazz);
    }

    /**
     * 可以传入配置 如超时时间等
     *
     * @param clazz
     * @param builder
     * @param <T>
     * @return
     */
    public <T> T wrap(Class clazz, ConsumerConfig.ConsumerConfigBuilder builder) {
        return (T) builder.setClient(client).build().refer(clazz);
    }

    /**
     * 泛化调用
     * params 是参数的json字符串形式
     * 在不依赖jar的情况下即可调用
     */
    public String invoke(String className, String methodName, String[] types, String[] params) {
        RpcInvocation invocation = new RpcInvocation();
        invocation.setMethodName(Constants.$INVOKE);
        invocation.setClassName(className);
        invocation.setParameterTypeNames(types);
        String[] params1 = new String[params.length + 1];
        params1[0] = methodName;
        System.arraycopy(params, 0, params1, 1, params.length);
        invocation.setArguments(params1);

        Invoker invoker = new RpcConsumerInvoker(client);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.CONSUMER, Sets.newHashSet());

        Result result = wrapperInvoker.invoke(invocation);
        if (result.hasException()) {
            throw new RpcException(result.getException());
        }
        return result.getValue().toString();
    }

    /**
     * echo调用
     * 测试服务是否可连通
     */
    public String echo(String className, String str) {
        RpcInvocation invocation = new RpcInvocation();
        invocation.setMethodName(Constants.$ECHO);
        invocation.setClassName(className);
        invocation.setParameterTypeNames(new String[]{ReflectUtils.getName(String.class)});
        invocation.setArguments(new String[]{str});
        RpcConsumerInvoker invoker = new RpcConsumerInvoker(client);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.CONSUMER, Sets.newHashSet());
        Result result = wrapperInvoker.invoke(invocation);
        if (result.hasException()) {
            throw new RuntimeException(result.getException());
        }
        return result.getValue().toString();
    }


    /**
     * 热部署
     * 动态修改class bytecode
     */
    public String deploy(String className, String classPath, String token) throws IOException {
        RpcInvocation invocation = new RpcInvocation();
        invocation.setMethodName(Constants.$HOT_DEPLOY);
        invocation.setClassName(className);

        String[] paramNames = new String[3];
        paramNames[0] = ReflectUtils.getName(String.class);
        paramNames[1] = ReflectUtils.getName(String.class);
        paramNames[2] = ReflectUtils.getName(byte[].class);
        invocation.setParameterTypeNames(paramNames);
        byte[] classData = Files.readAllBytes(Paths.get(classPath));
        Object[] params = new Object[]{className, token, classData};
        invocation.setArguments(params);
        RpcConsumerInvoker invoker = new RpcConsumerInvoker(client);
        Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.CONSUMER, Sets.newHashSet());
        Result result = wrapperInvoker.invoke(invocation);
        if (result.hasException()) {
            throw new RuntimeException(result.getException());
        }
        return result.getValue().toString();
    }

    public void close() {
        logger.info("==================>close");
        this.client.close();
    }

    public IClient getClient() {
        return client;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/monitor/MonitorService.java<|end_filename|>
package com.colobu.rpcx.monitor;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.rpc.URL;

import java.util.List;

/**
 * Created by <EMAIL>.
 */
public interface MonitorService {
    
    String APPLICATION = "application";
    
    String INTERFACE = "interface";

    String METHOD = "method";

    String GROUP = "group";

    String VERSION = "version";

    String CONSUMER = "consumer";

    String PROVIDER = "provider";
    
    String TIMESTAMP = "timestamp";

    String SUCCESS = "success";

    String FAILURE = "failure";
    
    String INPUT = Constants.INPUT_KEY;

    String OUTPUT = Constants.OUTPUT_KEY;

    String ELAPSED = "elapsed";

    String CONCURRENT = "concurrent";

    String MAX_INPUT = "max.input";

    String MAX_OUTPUT = "max.output";

    String MAX_ELAPSED = "max.elapsed";

    String MAX_CONCURRENT = "max.concurrent";

    /**
     * 监控数据采集.
     * 1. 支持调用次数统计：count://host/interface?application=foo&method=foo&provider=10.20.153.11:20880&success=12&failure=2&elapsed=135423423
     * 1.1 host,application,interface,group,version,method 记录监控来源主机，应用，接口，方法信息。
     * 1.2 如果是消费者发送的数据，加上provider地址参数，反之，加上来源consumer地址参数。
     * 1.3 success,faulure,elapsed 记录距上次采集，调用的成功次数，失败次数，成功调用总耗时，平均时间将用总耗时除以成功次数。
     * 
     * @param statistics
     */
    void collect(URL statistics);

    /**
     * 监控数据查询. 
     * 1. 支持按天查询：count://host/interface?application=foo&method=foo&side=provider&view=chart&date=2012-07-03
     * 1.1 host,application,interface,group,version,method 查询主机，应用，接口，方法的匹配条件，缺失的条件的表示全部，host用0.0.0.0表示全部。
     * 1.2 side=consumer,provider 查询由调用的哪一端采集的数据，缺省为都查询。
     * 1.3 缺省为view=summary，返回全天汇总信息，支持view=chart表示返回全天趋势图表图片的URL地址，可以进接嵌入其它系统的页面上展示。
     * 1.4 date=2012-07-03 指定查询数据的日期，缺省为当天。
     * 
     * @param query
     * @return statistics
     */
    List<URL> lookup(URL query);

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/ResponseFuture.java<|end_filename|>
package com.colobu.rpcx.netty;


import com.colobu.rpcx.common.InvokeCallback;
import com.colobu.rpcx.common.SemaphoreReleaseOnlyOnce;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.protocol.MessageStatusType;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.rpc.HessianUtils;
import com.colobu.rpcx.rpc.RpcException;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @author <EMAIL>
 */
public class ResponseFuture<T> {
    private final int opaque;
    private final long timeoutMillis;
    private final InvokeCallback invokeCallback;
    private final long beginTimestamp = System.currentTimeMillis();
    private final CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 保证只会被释放一次
     */
    private final SemaphoreReleaseOnlyOnce once;

    private final AtomicBoolean executeCallbackOnlyOnce = new AtomicBoolean(false);
    private volatile RemotingCommand responseCommand;
    private volatile boolean sendRequestOK = true;
    private volatile Throwable cause;


    public ResponseFuture(int opaque, long timeoutMillis, InvokeCallback invokeCallback,
                          SemaphoreReleaseOnlyOnce once) {
        this.opaque = opaque;
        this.timeoutMillis = timeoutMillis;
        this.invokeCallback = invokeCallback;
        this.once = once;
    }


    public void executeInvokeCallback() {
        if (invokeCallback != null) {
            if (this.executeCallbackOnlyOnce.compareAndSet(false, true)) {
                invokeCallback.operationComplete(this);
            }
        }
    }


    public void release() {
        if (this.once != null) {
            this.once.release();
        }
    }


    public boolean isTimeout() {
        long diff = System.currentTimeMillis() - this.beginTimestamp;
        return diff > this.timeoutMillis;
    }


    public RemotingCommand waitResponse(final long timeoutMillis) throws InterruptedException {
        this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
        return this.responseCommand;
    }

    /**
     * 放入结果就是coutDownLatch 执行countDown  waitResponse就不会再阻塞
     *
     * @param responseCommand
     */
    public void putResponse(final RemotingCommand responseCommand) {
        this.responseCommand = responseCommand;
        this.countDownLatch.countDown();
    }


    public long getBeginTimestamp() {
        return beginTimestamp;
    }


    public boolean isSendRequestOK() {
        return sendRequestOK;
    }


    public void setSendRequestOK(boolean sendRequestOK) {
        this.sendRequestOK = sendRequestOK;
    }


    public long getTimeoutMillis() {
        return timeoutMillis;
    }


    public InvokeCallback getInvokeCallback() {
        return invokeCallback;
    }


    public Throwable getCause() {
        return cause;
    }


    public void setCause(Throwable cause) {
        this.cause = cause;
    }


    public RemotingCommand getResponseCommand() {
        return responseCommand;
    }


    public void setResponseCommand(RemotingCommand responseCommand) {
        this.responseCommand = responseCommand;
    }


    public int getOpaque() {
        return opaque;
    }


    @Override
    public String toString() {
        return "ResponseFuture [responseCommand=" + responseCommand + ", sendRequestOK=" + sendRequestOK
                + ", cause=" + cause + ", opaque=" + opaque + ", timeoutMillis=" + timeoutMillis
                + ", invokeCallback=" + invokeCallback + ", beginTimestamp=" + beginTimestamp
                + ", countDownLatch=" + countDownLatch + "]";
    }

    public T get(long timeoutMillis) throws InterruptedException {
        RemotingCommand res = this.waitResponse(timeoutMillis);
        //如果有错误,则输出错误异常
        if (res.getMessage().getMessageStatusType().equals(MessageStatusType.Error)) {
            throw new RpcException(res.getMessage().getMetadata().get(Constants.RPCX_ERROR_MESSAGE), res.getMessage().getMetadata().get(Constants.RPCX_ERROR_CODE));
        } else {
            byte[] data = res.getMessage().payload;
            return (T) HessianUtils.read(data);
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/cache/LruCacheFactory.java<|end_filename|>
package com.colobu.rpcx.cache;


import com.colobu.rpcx.cache.impl.LruCache;
import com.colobu.rpcx.rpc.URL;

/**
 * @author <EMAIL>
 */
public class LruCacheFactory extends AbstractCacheFactory {

    @Override
    protected Cache createCache(URL url) {
        return new LruCache();
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/utils/PathStatus.java<|end_filename|>
package com.colobu.rpcx.utils;

import com.colobu.rpcx.common.Pair;

/**
 * @author zhangzhiyong
 * @date 2018/7/4
 */
public class PathStatus {

    private String type;
    private Pair<String,String> value;
    private String path;

    private boolean stop;

    public PathStatus(String type, Pair<String,String> value, String path) {
        this.type = type;
        this.value = value;
        this.path = path;
    }


    public PathStatus(boolean stop) {
        this.stop = stop;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public Pair<String,String> getValue() {
        return value;
    }

    public void setValue(Pair<String,String> value) {
        this.value = value;
    }

    public String getPath() {
        return path;
    }

    public void setPath(String path) {
        this.path = path;
    }

    @Override
    public String toString() {
        return "PathStatus{" +
                "type='" + type + '\'' +
                ", value='" + value + '\'' +
                '}';
    }

    public boolean isStop() {
        return stop;
    }

    public void setStop(boolean stop) {
        this.stop = stop;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/utils/ZkClient.java<|end_filename|>
package com.colobu.rpcx.utils;

import com.colobu.rpcx.common.Config;
import com.colobu.rpcx.common.Pair;
import com.colobu.rpcx.rpc.URL;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * Created by zhangzhiyong on 2018/7/4.
 */
public class ZkClient {

    private static final Logger logger = LoggerFactory.getLogger(ZkClient.class);

    private CuratorFramework client;


    private CopyOnWriteArrayList<PathChildrenCache> list = new CopyOnWriteArrayList<>();

    private AtomicBoolean stop = new AtomicBoolean(false);


    private ZkClient() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String cs = Config.ins().get("rpcx.zk.connect.string");
        client =
                CuratorFrameworkFactory.builder()
                        .connectString(cs)
                        .sessionTimeoutMs(5000)
                        .connectionTimeoutMs(5000)
                        .retryPolicy(retryPolicy)
                        .build();
        client.start();
    }

    private static final class LazyHolder {
        private static final ZkClient ins = new ZkClient();
    }

    public static ZkClient ins() {
        return LazyHolder.ins;
    }


    public Set<Pair<String, String>> get(String basePath, String serviceName) {
        try {
            Set<String> set = client.getChildren().forPath(basePath + serviceName).stream().collect(Collectors.toSet());
            Set<Pair<String, String>> pairSet = set.stream().map(it -> {
                String data = "";
                try {
                    data = new String(client.getData().forPath(basePath + serviceName + "/" + it));
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return Pair.of(it, data);
            }).collect(Collectors.toSet());
            return pairSet;
        } catch (Exception e) {
            logger.error("get error:{}", e.getMessage());
        }
        return new HashSet<>();
    }


    ///youpin/services/Arith/tcp@0.0.0.0:8976"
    public void create(String basePath, Set<String> serviceNames, String addr) {
        serviceNames.forEach(u -> {
            URL url = URL.valueOf(u);
            //& 为了兼容  rpcx(golang)
//            String path = basePath + url.getPath() + "/tcp@" + addr + "&" + url.toParameterString();
            String path = basePath + url.getPath() + "/tcp@" + addr;
            try {
                if (client.checkExists().forPath(path) == null) {
                    client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path, url.toParameterString().getBytes());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }


    //监控变化
    public void watch(LinkedBlockingQueue<PathStatus> queue, final String p) throws Exception {
        if (this.stop.get() == true) {
            return;
        }
        logger.info("---------->watch path:{}", p);
        PathChildrenCache cache = new PathChildrenCache(this.client, p, true);
        list.add(cache);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((cf, event) -> {
            switch (event.getType()) {
                case CHILD_ADDED:
                    String path = event.getData().getPath();
                    String param = new String(event.getData().getData());
                    String addr = path.split("@")[1];
                    logger.info("------->CHILD_ADDED :{}:{}", path, addr);
                    queue.offer(new PathStatus("CHILD_ADDED", Pair.of(addr, param), p));
                    break;
                case CHILD_UPDATED:
                    String updatePath = event.getData().getPath();
                    logger.info("------>CHILD_UPDATED :" + event.getData().getPath());
                    String updateParam = new String(event.getData().getData());
                    String updateAddr = updatePath.split("@")[1];
                    queue.offer(new PathStatus("CHILD_UPDATED", Pair.of(updateAddr, updateParam), p));
                    break;
                case CHILD_REMOVED:
                    String path2 = event.getData().getPath();
                    String addr2 = path2.split("@")[1];
                    logger.info("------>CHILD_REMOVED :{}:{}" + path2, addr2);
                    queue.offer(new PathStatus("CHILD_REMOVED", Pair.of(addr2, ""), p));
                    break;
                default:
                    break;
            }
        });
    }

    public void close() {
        this.stop.compareAndSet(false, true);
        this.list.forEach(it -> {
            try {
                it.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

        this.client.close();
    }

}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/http/RpcxHttpClient.java<|end_filename|>
package com.colobu.rpcx.client.http;

import com.colobu.rpcx.common.Pair;
import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.rpc.Invocation;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.google.gson.Gson;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * @author <EMAIL>
 */
public class RpcxHttpClient {

    private static final Logger logger = LoggerFactory.getLogger(RpcxHttpClient.class);

    /**
     * http 调用
     *
     * @param url
     * @param service
     * @param method
     * @param params  key=类型  value=json后的值
     * @return
     */
    public static String execute(String url, String service, String method, Pair<String, String>... params) {
        RequestConfig defaultRequestConfig = RequestConfig.custom()
                .setSocketTimeout(1000)
                .setConnectTimeout(1000)
                .setConnectionRequestTimeout(1000)
                .build();

        try (CloseableHttpClient client = HttpClients.custom().setDefaultRequestConfig(defaultRequestConfig).build()) {
            HttpPost post = new HttpPost(url);
            post.setHeader("X-RPCX-ServicePath", service);
            post.setHeader("X-RPCX-ServiceMethod", method);
            post.setHeader("connection", "close");

            Gson gson = new Gson();
            Invocation invocation = new RpcInvocation();
            String[] parameterTypeNames = new String[params.length];
            Object[] arguments = new Object[params.length];
            IntStream.range(0, params.length).forEach(index -> {
                Pair<String, String> p = params[index];
                parameterTypeNames[index] = p.getObject1();
                arguments[index] = p.getObject2();
            });

            invocation.setParameterTypeNames(parameterTypeNames);
            invocation.setArguments(arguments);

            URL u = new URL("rpcx", "", 0);

            ((RpcInvocation) invocation).setClassName(service);
            ((RpcInvocation) invocation).setMethodName(method);

            u.setServiceInterface(invocation.getClassName() + "." + invocation.getMethodName());
            String _params = Stream.of(invocation.getArguments()).map(it -> gson.toJson(it)).collect(Collectors.joining(","));
            u.setPath(invocation.getClassName() + "." + invocation.getMethodName() + "(" + _params + ")");
            ((RpcInvocation) invocation).setUrl(u);

            byte[] payload = new Gson().toJson(invocation).getBytes();

            ByteArrayEntity entriy = new ByteArrayEntity(payload, ContentType.DEFAULT_BINARY);
            post.setEntity(entriy);
            HttpResponse res = client.execute(post);
            HttpEntity resEntity = res.getEntity();
            byte[] data = EntityUtils.toByteArray(resEntity);
            Message message = gson.fromJson(new String(data), Message.class);
            return new String(message.payload);
        } catch (Exception ex) {
            logger.info("execute error:{}", ex.getMessage());
            throw new RpcException(ex);
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/Node.java<|end_filename|>
package com.colobu.rpcx.rpc;

/**
 * Created by <EMAIL>.
 */
public interface Node {

    URL getUrl();

    void setUrl(URL url);

    boolean isAvailable();

    void destroy();

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/cache/Cache.java<|end_filename|>
package com.colobu.rpcx.cache;

/**
 * @author <EMAIL>
 */
public interface Cache {

    void put(Object key, Object value);

    Object get(Object key);

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/RpcInvocation.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.fail.FailType;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.protocol.SerializeType;
import com.colobu.rpcx.rpc.Invocation;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.selector.SelectMode;
import io.netty.channel.Channel;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;


/**
 * @author <EMAIL>
 */
public class RpcInvocation implements Invocation, Serializable {

    private String className;

    private String methodName;

    private String group = "";

    private Class<?>[] parameterTypes;

    private Class<?> resultType;

    public String[] parameterTypeNames;

    private Object[] arguments;

    private Map<String, String> attachments = new HashMap<>(1);

    private transient Invoker<?> invoker;

    public URL url;

    private long timeOut = TimeUnit.SECONDS.toMillis(1);

    private int retryNum = 1;

    private String sendType = Constants.SYNC_KEY;

    public LanguageCode languageCode = LanguageCode.JAVA;


    private SerializeType serializeType = SerializeType.SerializeNone;


    private FailType failType = FailType.FailFast;

    private SelectMode selectMode = SelectMode.RandomSelect;

    private transient byte[] payload;

    public RpcInvocation() {
    }


    @Override
    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    @Override
    public Class<?>[] getParameterTypes() {
        return parameterTypes;
    }

    @Override
    public String[] getParameterTypeNames() {
        return this.parameterTypeNames;
    }

    public void setParameterTypes(Class<?>[] parameterTypes) {
        this.parameterTypes = parameterTypes;
    }

    @Override
    public Object[] getArguments() {
        return arguments;
    }

    @Override
    public void setArguments(Object[] arguments) {
        this.arguments = arguments;
    }

    @Override
    public Map<String, String> getAttachments() {
        return attachments;
    }

    @Override
    public String getAttachment(String key) {
        return this.attachments.get(key);
    }

    @Override
    public String getAttachment(String key, String defaultValue) {
        String value = this.attachments.get(key);
        if (null == value) {
            return defaultValue;
        }
        return value;
    }

    public void setAttachments(Map<String, String> attachments) {
        this.attachments = attachments;
    }

    @Override
    public Invoker<?> getInvoker() {
        return invoker;
    }

    public void setInvoker(Invoker<?> invoker) {
        this.invoker = invoker;
    }

    @Override
    public String getClassName() {
        return className;
    }

    public void setClassName(String className) {
        this.className = className;
    }

    @Override
    public Class<?> getResultType() {
        return resultType;
    }

    public void setResultType(Class<?> resultType) {
        this.resultType = resultType;
    }

    @Override
    public void setParameterTypeNames(String[] parameterTypeNames) {
        this.parameterTypeNames = parameterTypeNames;
    }

    public long getTimeOut() {
        return timeOut;
    }

    public void setTimeOut(long timeOut) {
        this.timeOut = timeOut;
    }

    public int getRetryNum() {
        return retryNum;
    }

    public void setRetryNum(int retryNum) {
        this.retryNum = retryNum;
    }

    public URL getUrl() {
        return url;
    }

    public void setUrl(URL url) {
        this.url = url;
    }

    public String getSendType() {
        return sendType;
    }

    public void setSendType(String sendType) {
        this.sendType = sendType;
    }

    public FailType getFailType() {
        return failType;
    }

    public void setFailType(FailType failType) {
        this.failType = failType;
    }


    public SelectMode getSelectMode() {
        return selectMode;
    }

    public void setSelectMode(SelectMode selectMode) {
        this.selectMode = selectMode;
    }

    public LanguageCode getLanguageCode() {
        return languageCode;
    }

    public void setLanguageCode(LanguageCode languageCode) {
        this.languageCode = languageCode;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public byte[] getPayload() {
        return payload;
    }

    public void setPayload(byte[] payload) {
        this.payload = payload;
    }

    public SerializeType getSerializeType() {
        return serializeType;
    }

    public void setSerializeType(SerializeType serializeType) {
        this.serializeType = serializeType;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/RemotingCommandType.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum RemotingCommandType {

    /**
     * 请求命令
     */
    REQUEST_COMMAND,

    /**
     * 应答命令
     */
    RESPONSE_COMMAND
}


<|start_filename|>demo/api/src/main/java/com/colobu/rpcx/service/ITest2Service.java<|end_filename|>
package com.colobu.rpcx.service;


import com.colobu.rpcx.rpc.annotation.Consumer;

@Consumer(impl = "com.colobu.rpcx.service.Test2Service")
public interface ITest2Service {

    int num();

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/annotation/RpcFilter.java<|end_filename|>
package com.colobu.rpcx.rpc.annotation;


import java.lang.annotation.*;


/**
 * Created by <EMAIL>.
 */
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
@Documented
public @interface RpcFilter {

    //数字越小的越先执行
    int order() default 0;

    String[] group() default {};

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/Protocol.java<|end_filename|>
package com.colobu.rpcx.rpc;

/**
 * Created by <EMAIL>.
 */
public class Protocol {
}


<|start_filename|>demo/demoserver2/src/main/java/com/colobu/rpcx/bootstrap/Server2Bootstrap.java<|end_filename|>
package com.colobu.rpcx.bootstrap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.web.bind.annotation.RestController;


/**
 * @author <EMAIL>
 */
@EnableAutoConfiguration
@ComponentScan(basePackages = {"com.colobu.rpcx"})
@RestController
public class Server2Bootstrap {

    private static final Logger logger = LoggerFactory.getLogger(Server2Bootstrap.class);

    public static void main(String... args) {
        logger.info("Server2Bootstrap start");
        try {
            SpringApplication.run(Server2Bootstrap.class, args);
        } catch (Throwable ex) {
            logger.error(ex.getMessage(), ex);
            System.exit(0);
        }
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/TpsLimitFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.tps.DefaultTPSLimiter;
import com.colobu.rpcx.tps.TPSLimiter;

/**
 * Created by <EMAIL>.
 */
@RpcFilter(group = {Constants.PROVIDER})
public class TpsLimitFilter implements Filter {

    private static final TPSLimiter tpsLimiter = new DefaultTPSLimiter();

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        if (invocation.getUrl().getParameter(Constants.TPS_LIMIT_RATE_KEY,"-1").equals("-1")) {
            return invoker.invoke(invocation);
        }

        if (!tpsLimiter.isAllowable(invoker.getUrl(), invocation)) {
            throw new RpcException(
                    new StringBuilder(64)
                            .append("Failed to invoke service ")
                            .append(invoker.getInterface().getName())
                            .append(".")
                            .append(invocation.getMethodName())
                            .append(" because exceed max service tps.")
                            .toString());
        }

        return invoker.invoke(invocation);
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/fail/FailType.java<|end_filename|>
package com.colobu.rpcx.fail;

/**
 * @author <EMAIL>
 */
public enum FailType {
    /**
     * 立刻返回
     */
    FailFast,
    /**
     * 换一个服务节点
     */
    FailOver,
    /**
     * 在当前节点重试
     */
    FailTry,
}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/service/DemoServiceImpl.java<|end_filename|>
package com.colobu.rpcx.service;


import com.colobu.rpcx.rpc.annotation.Provider;
import org.springframework.stereotype.Service;

/**
 * @author <EMAIL>
 */
@Service
@Provider
public class DemoServiceImpl implements DemoService {


    public String test(String str) {
        return "test" + "," + str;
    }
}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/demo/Server.java<|end_filename|>
package com.colobu.rpcx.demo;

import com.colobu.rpcx.register.IServiceRegister;
import com.colobu.rpcx.server.NettyServer;
import com.colobu.rpcx.server.ZkServiceRegister;


/**
 * @author <EMAIL>
 */
public class Server {

    public static void main(String... args) {
        NettyServer server = new NettyServer();
        server.start();
        IServiceRegister reg = new ZkServiceRegister("/youpin/services/", server.getAddr() + ":" + server.getPort(), "com.colobu", null);
        reg.register();
        reg.start();
        server.await();
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/monitor/impl/Statistics.java<|end_filename|>
package com.colobu.rpcx.monitor.impl;


import com.colobu.rpcx.monitor.MonitorService;
import com.colobu.rpcx.rpc.URL;

import java.io.Serializable;


/**
 * Created by <EMAIL>.
 */
public class Statistics implements Serializable {
    
    private static final long serialVersionUID = -6921183057683641441L;
    
    private URL url;
    
    private String application;
    
    private String service;

    private String method;
    
    private String group;

    private String version;
    
    private String client;
    
    private String server;

    public Statistics(URL url) {
        this.url = url;
        this.application = url.getParameter(MonitorService.APPLICATION);
        this.service = url.getParameter(MonitorService.INTERFACE);
        this.method = url.getParameter(MonitorService.METHOD);
        this.group = url.getParameter(MonitorService.GROUP);
        this.version = url.getParameter(MonitorService.VERSION);
        this.client = url.getParameter(MonitorService.CONSUMER, url.getAddress());
        this.server = url.getParameter(MonitorService.PROVIDER, url.getAddress());
    }

    public URL getUrl() {
        return url;
    }

    public void setUrl(URL url) {
        this.url = url;
    }

    public String getApplication() {
        return application;
    }

    public Statistics setApplication(String application) {
        this.application = application;
        return this;
    }

    public String getService() {
        return service;
    }

    public Statistics setService(String service) {
        this.service = service;
        return this;
    }

    public String getGroup() {
        return group;
    }

    public void setGroup(String group) {
        this.group = group;
    }

    public String getVersion() {
        return version;
    }

    public void setVersion(String version) {
        this.version = version;
    }

    public String getMethod() {
        return method;
    }

    public Statistics setMethod(String method) {
        this.method = method;
        return this;
    }

    public String getClient() {
        return client;
    }

    public Statistics setClient(String client) {
        this.client = client;
        return this;
    }

    public String getServer() {
        return server;
    }

    public Statistics setServer(String server) {
        this.server = server;
        return this;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((application == null) ? 0 : application.hashCode());
        result = prime * result + ((client == null) ? 0 : client.hashCode());
        result = prime * result + ((group == null) ? 0 : group.hashCode());
        result = prime * result + ((method == null) ? 0 : method.hashCode());
        result = prime * result + ((server == null) ? 0 : server.hashCode());
        result = prime * result + ((service == null) ? 0 : service.hashCode());
        result = prime * result + ((version == null) ? 0 : version.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        Statistics other = (Statistics) obj;
        if (application == null) {
            if (other.application != null) {
                return false;
            }
        } else if (!application.equals(other.application)) {
            return false;
        }
        if (client == null) {
            if (other.client != null) {
                return false;
            }
        } else if (!client.equals(other.client)) {
            return false;
        }
        if (group == null) {
            if (other.group != null) {
                return false;
            }
        } else if (!group.equals(other.group)) {
            return false;
        }
        if (method == null) {
            if (other.method != null) {
                return false;
            }
        } else if (!method.equals(other.method)) {
            return false;
        }
        if (server == null) {
            if (other.server != null) {
                return false;
            }
        } else if (!server.equals(other.server)) {
            return false;
        }
        if (service == null) {
            if (other.service != null) {
                return false;
            }
        } else if (!service.equals(other.service)) {
            return false;
        }
        if (version == null) {
            if (other.version != null) {
                return false;
            }
        } else if (!version.equals(other.version)) {
            return false;
        }
        return true;
    }

    @Override
    public String toString() {
        return url.toString();
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/RpcContext.java<|end_filename|>
package com.colobu.rpcx.rpc;


import com.colobu.rpcx.common.NetUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.netty.ResponseFuture;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;


/**
 * @author <EMAIL>
 */
public class RpcContext {

    private static final ThreadLocal<RpcContext> LOCAL = ThreadLocal.withInitial(() -> new RpcContext());

    public static RpcContext getContext() {
        return LOCAL.get();
    }

    public static void removeContext() {
        LOCAL.remove();
    }

    private ResponseFuture future;

    private List<URL> urls;

    private URL url;

    private String methodName;

    private Class<?>[] parameterTypes;

    private Object[] arguments;

    private InetSocketAddress localAddress;

    private InetSocketAddress remoteAddress;

    private final Map<String, String> attachments = new HashMap<>();

    private final Map<String, Object> values = new HashMap<>();

    private String serviceAddr;

    protected RpcContext() {
    }

    /**
     * is provider side.
     *
     * @return provider side.
     */
    public boolean isProviderSide() {
        URL url = getUrl();
        if (url == null) {
            return false;
        }
        InetSocketAddress address = getRemoteAddress();
        if (address == null) {
            return false;
        }
        String host;
        if (address.getAddress() == null) {
            host = address.getHostName();
        } else {
            host = address.getAddress().getHostAddress();
        }
        return url.getPort() != address.getPort() ||
                !NetUtils.filterLocalHost(url.getIp()).equals(NetUtils.filterLocalHost(host));
    }

    /**
     * is consumer side.
     *
     * @return consumer side.
     */
    public boolean isConsumerSide() {
        URL url = getUrl();
        if (url == null) {
            return false;
        }
        InetSocketAddress address = getRemoteAddress();
        if (address == null) {
            return false;
        }
        String host;
        if (address.getAddress() == null) {
            host = address.getHostName();
        } else {
            host = address.getAddress().getHostAddress();
        }
        return url.getPort() == address.getPort() &&
                NetUtils.filterLocalHost(url.getIp()).equals(NetUtils.filterLocalHost(host));
    }

    /**
     * get future.
     *
     * @return future
     */
    public ResponseFuture getFuture() {
        return this.future;
    }

    /**
     * set future.
     *
     * @param future
     */
    public void setFuture(ResponseFuture future) {
        this.future = future;
    }

    public List<URL> getUrls() {
        return urls == null && url != null ? (List<URL>) Arrays.asList(url) : urls;
    }

    public void setUrls(List<URL> urls) {
        this.urls = urls;
    }

    public URL getUrl() {
        return url;
    }

    public void setUrl(URL url) {
        this.url = url;
    }

    /**
     * get method name.
     *
     * @return method name.
     */
    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    /**
     * get parameter types.
     *
     * @serial
     */
    public Class<?>[] getParameterTypes() {
        return parameterTypes;
    }

    public void setParameterTypes(Class<?>[] parameterTypes) {
        this.parameterTypes = parameterTypes;
    }

    /**
     * get arguments.
     *
     * @return arguments.
     */
    public Object[] getArguments() {
        return arguments;
    }

    public void setArguments(Object[] arguments) {
        this.arguments = arguments;
    }

    /**
     * set local address.
     *
     * @param address
     * @return context
     */
    public RpcContext setLocalAddress(InetSocketAddress address) {
        this.localAddress = address;
        return this;
    }

    /**
     * set local address.
     *
     * @param host
     * @param port
     * @return context
     */
    public RpcContext setLocalAddress(String host, int port) {
        if (port < 0) {
            port = 0;
        }
        this.localAddress = InetSocketAddress.createUnresolved(host, port);
        return this;
    }

    /**
     * get local address.
     *
     * @return local address
     */
    public InetSocketAddress getLocalAddress() {
        return localAddress;
    }

    public String getLocalAddressString() {
        return getLocalHost() + ":" + getLocalPort();
    }

    /**
     * get local host name.
     *
     * @return local host name
     */
    public String getLocalHostName() {
        String host = localAddress == null ? null : localAddress.getHostName();
        if (host == null || host.length() == 0) {
            return getLocalHost();
        }
        return host;
    }

    /**
     * set remote address.
     *
     * @param address
     * @return context
     */
    public RpcContext setRemoteAddress(InetSocketAddress address) {
        this.remoteAddress = address;
        return this;
    }

    /**
     * set remote address.
     *
     * @param host
     * @param port
     * @return context
     */
    public RpcContext setRemoteAddress(String host, int port) {
        if (port < 0) {
            port = 0;
        }
        this.remoteAddress = InetSocketAddress.createUnresolved(host, port);
        return this;
    }

    /**
     * get remote address.
     *
     * @return remote address
     */
    public InetSocketAddress getRemoteAddress() {
        return remoteAddress;
    }

    /**
     * get remote address string.
     *
     * @return remote address string.
     */
    public String getRemoteAddressString() {
        return getRemoteHost() + ":" + getRemotePort();
    }

    /**
     * get remote host name.
     *
     * @return remote host name
     */
    public String getRemoteHostName() {
        return remoteAddress == null ? null : remoteAddress.getHostName();
    }

    /**
     * get local host.
     *
     * @return local host
     */
    public String getLocalHost() {
        String host = localAddress == null ? null :
                localAddress.getAddress() == null ? localAddress.getHostName()
                        : NetUtils.filterLocalHost(localAddress.getAddress().getHostAddress());
        if (host == null || host.length() == 0) {
            return NetUtils.getLocalHost();
        }
        return host;
    }

    /**
     * get local port.
     *
     * @return port
     */
    public int getLocalPort() {
        return localAddress == null ? 0 : localAddress.getPort();
    }

    /**
     * get remote host.
     *
     * @return remote host
     */
    public String getRemoteHost() {
        return remoteAddress == null ? null :
                remoteAddress.getAddress() == null ? remoteAddress.getHostName()
                        : NetUtils.filterLocalHost(remoteAddress.getAddress().getHostAddress());
    }

    /**
     * get remote port.
     *
     * @return remote port
     */
    public int getRemotePort() {
        return remoteAddress == null ? 0 : remoteAddress.getPort();
    }

    /**
     * get attachment.
     *
     * @param key
     * @return attachment
     */
    public String getAttachment(String key) {
        return attachments.get(key);
    }

    /**
     * set attachment.
     *
     * @param key
     * @param value
     * @return context
     */
    public RpcContext setAttachment(String key, String value) {
        if (value == null) {
            attachments.remove(key);
        } else {
            attachments.put(key, value);
        }
        return this;
    }

    /**
     * remove attachment.
     *
     * @param key
     * @return context
     */
    public RpcContext removeAttachment(String key) {
        attachments.remove(key);
        return this;
    }

    /**
     * get attachments.
     *
     * @return attachments
     */
    public Map<String, String> getAttachments() {
        return attachments;
    }

    /**
     * set attachments
     *
     * @param attachment
     * @return context
     */
    public RpcContext setAttachments(Map<String, String> attachment) {
        this.attachments.clear();
        if (attachment != null && attachment.size() > 0) {
            this.attachments.putAll(attachment);
        }
        return this;
    }

    public void clearAttachments() {
        this.attachments.clear();
    }

    /**
     * get values.
     *
     * @return values
     */
    public Map<String, Object> get() {
        return values;
    }

    public RpcContext set(String key, Object value) {
        if (value == null) {
            values.remove(key);
        } else {
            values.put(key, value);
        }
        return this;
    }

    public RpcContext remove(String key) {
        values.remove(key);
        return this;
    }

    public Object get(String key) {
        return values.get(key);
    }


    /**
     * @deprecated Replace to isProviderSide()
     */
    @Deprecated
    public boolean isServerSide() {
        return isProviderSide();
    }

    /**
     * @deprecated Replace to isConsumerSide()
     */
    @Deprecated
    public boolean isClientSide() {
        return isConsumerSide();
    }


    public String getServiceAddr() {
        return serviceAddr;
    }

    public void setServiceAddr(String serviceAddr) {
        this.serviceAddr = serviceAddr;
    }
}

<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/Message.java<|end_filename|>
package com.colobu.rpcx.protocol;

import com.colobu.rpcx.common.Bytes;
import com.colobu.rpcx.rpc.RpcException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.Data;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.zip.GZIPOutputStream;

/**
 * @author <EMAIL>
 */
@Data
public class Message {

    public static byte magicNumber = 0x08;

    public byte[] header;
    /**
     * java-->className
     */
    public String servicePath;
    /**
     * java-->methodName
     */
    public String serviceMethod;

    public Map<String, String> metadata;

    public byte[] payload;

    public Message(String servicePath, String serviceMethod) {
        this();
        this.servicePath = servicePath;
        this.serviceMethod = serviceMethod;
    }

    public Message() {
        header = new byte[12];
        header[0] = magicNumber;
        payload = new byte[]{};
        this.metadata = new HashMap<>(2);
    }


    public Message(String servicePath, String serviceMethod, MessageType messageType, long seq) {
        this(servicePath, serviceMethod);
        this.setMessageType(messageType);
        this.setSeq(seq);
    }

    public Message(String servicePath, String serviceMethod, Map<String, String> metadata) {
        this(servicePath, serviceMethod);
        this.metadata = metadata;
    }

    public Message(String servicePath, String serviceMethod, Map<String, String> metadata, byte[] payload) {
        this(servicePath, serviceMethod, metadata);
        this.payload = payload;
    }


    public byte getVersion() {
        return header[1];
    }

    public void setVersion(byte version) {
        header[1] = version;
    }

    public MessageType getMessageType() {
        if ((header[2] & 0x80) == 0) {
            return MessageType.Request;
        }

        return MessageType.Response;
    }

    public void setMessageType(MessageType mt) {
        if (mt == MessageType.Request) {
            header[2] &= ~0x80;
        } else {
            header[2] |= 0x80;
        }
    }

    public void setMessageType(byte type) {
        header[2] = type;
    }


    public boolean isHeartbeat() {
        return (header[2] & 0x40) != 0;
    }

    public void setHeartbeat(boolean heartbeat) {
        if (heartbeat) {
            header[2] |= 0x40;
        } else {
            header[2] &= ~0x40;
        }
    }

    public boolean isOneway() {
        return (header[2] & 0x20) != 0;
    }

    public void setOneway(boolean oneway) {
        if (oneway) {
            header[2] |= 0x20;
        } else {
            header[2] &= ~0x20;
        }
    }


    public CompressType getCompressType() {
        int v = (header[2] & 0x1C) >> 2;
        return CompressType.getValue(v);
    }

    public void setCompressType(CompressType ct) {
        int v = ct.value();
        header[2] &= ~0x1C;
        header[2] |= (v << 2) & 0x1C;
    }

    public MessageStatusType getMessageStatusType() {
        int v = header[2] & 0x03;
        return MessageStatusType.getValue(v);
    }

    //如果返回结果有error,这里需要设置
    public void setMessageStatusType(MessageStatusType mst) {
        int v = mst.value();
        header[2] &= ~0x03;
        header[2] |= v & 0x03;
    }

    public SerializeType getSerializeType() {
        int v = (header[3] & 0xF0) >> 4;
        return SerializeType.getValue(v);
    }

    public void setSerializeType(SerializeType st) {
        int v = st.value();
        header[3] &= 0x0F;
        header[3] |= (v << 4) & 0xF0;
    }

    public long getSeq() {
        return Bytes.bytes2long(header, 4);
    }

    public void setSeq(long seq) {
        System.arraycopy(Bytes.long2bytes(seq), 0, header, 4, 8);
    }


    private byte[] compress() throws IOException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        GZIPOutputStream zipStream = new GZIPOutputStream(bos);
        zipStream.write(payload);
        return bos.toByteArray();
    }


    /**
     * 编码
     *
     * @return
     * @throws Exception
     */
    public byte[] encode() throws Exception {
        if (getCompressType() == CompressType.Gzip) {
            this.payload = compress();
        }

        byte[] spBytes = servicePath.getBytes();
        byte[] smBytes = serviceMethod.getBytes();
        ByteBuf metaBytes = encodeMetadata();

        int headLen = header.length;
        int bodyLen = spBytes.length + 4 + smBytes.length + 4 + metaBytes.capacity() + 4 + payload.length + 4;

        int capacity = headLen + 4 + bodyLen;
        ByteBuffer buffer = ByteBuffer.allocate(capacity);
        buffer.put(header);

        buffer.putInt(bodyLen);

        buffer.putInt(spBytes.length);
        buffer.put(spBytes);

        buffer.putInt(smBytes.length);
        buffer.put(smBytes);

        buffer.putInt(metaBytes.capacity());
        buffer.put(metaBytes.array());

        buffer.putInt(payload.length);
        buffer.put(payload);

        return buffer.array();
    }


    private ByteBuf encodeMetadata() {
        if (metadata.size() == 0) {
            return Unpooled.EMPTY_BUFFER;
        }

        ByteBuf buf = Unpooled.buffer(metadata.size() * 15);

        for (Map.Entry<String, String> entry : metadata.entrySet()) {
            String key = entry.getKey();
            byte[] keyBytes = key.getBytes();
            buf.writeInt(keyBytes.length);
            buf.writeBytes(keyBytes);

            String v = entry.getValue();
            if (null == v) {
                v = "null";
            }
            byte[] vBytes = v.getBytes();
            buf.writeInt(vBytes.length);
            buf.writeBytes(vBytes);
        }
        return buf.capacity(buf.writerIndex());
    }


    /**
     * 业务解码
     */
    public void decode(byte[] data) throws UnsupportedEncodingException {
        ByteBuffer buf = ByteBuffer.wrap(data);
        //servicePath
        int len = buf.getInt();
        byte[] b = new byte[len];
        buf.get(b);
        servicePath = new String(b, "UTF-8");
        //serviceMethod
        len = buf.getInt();
        b = new byte[len];
        buf.get(b);
        serviceMethod = new String(b, "UTF-8");
        //metadata
        len = buf.getInt();
        b = new byte[len];
        buf.get(b);
        decodeMetadata(b);
        //payload
        len = buf.getInt();
        byte[] payload = new byte[len];
        buf.get(payload);
        this.payload = payload;
    }

    private void decodeMetadata(byte[] b) throws UnsupportedEncodingException {
        int blen = b.length;
        if (blen > 8) {
            int len;
            int index = 0;
            byte[] data = null;
            for (; ; ) {
                if (blen - index < 4) {
                    break;
                }
                len = Bytes.bytes2int(b, index);
                index = index + 4;
                data = new byte[len];
                System.arraycopy(b, index, data, 0, len);
                String key = new String(data, "UTF-8");
                index = index + len;


                len = Bytes.bytes2int(b, index);
                index = index + 4;
                data = new byte[len];
                System.arraycopy(b, index, data, 0, len);
                String value = new String(data, "UTF-8");
                index = index + len;
                this.metadata.put(key, value);
            }
        }
    }


    public Map<String, String> getMetadata() {
        return metadata;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/FilterWrapper.java<|end_filename|>
package com.colobu.rpcx.filter;

import com.colobu.rpcx.common.Config;
import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.URL;
import com.colobu.rpcx.rpc.impl.RpcFilterFinder;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author <EMAIL>
 */
public class FilterWrapper {

    private static final Logger logger = LoggerFactory.getLogger(FilterWrapper.class);

    private List<Filter> providerFilters;
    private List<Filter> consumerFilters;


    private FilterWrapper() {
        this.providerFilters = this.findFilters("com.colobu", Constants.PROVIDER);
        this.consumerFilters = this.findFilters("com.colobu", Constants.CONSUMER);

        String p = Config.ins().get("rpcx.filter.package");
        //自己没有配置filter package 则不添加
        if (StringUtils.isNotEmpty(p)) {
            this.providerFilters.addAll(this.findFilters(p, Constants.PROVIDER));
            this.providerFilters.addAll(this.findFilters(p, Constants.CONSUMER));
        }
    }


    private List<Filter> findFilters(String packageName, String group) {
        RpcFilterFinder finder = new RpcFilterFinder(packageName, group);
        List<Class> list = finder.find();
        logger.info("################" + group + " " + list);
        return list.stream().map(it -> {
            try {
                return (Filter) it.newInstance();
            } catch (InstantiationException e) {
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
            return null;
        }).collect(Collectors.toList());
    }


    private static class LazyHolder {
        public static FilterWrapper ins = new FilterWrapper();
    }

    public static FilterWrapper ins() {
        return LazyHolder.ins;
    }

    /**
     * @param invoker
     * @param key
     * @param group
     * @param excludeFilters 不需要构建的filter
     * @param <T>
     * @return
     */
    public <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group, Set<String> excludeFilters) {
        Invoker<T> last = invoker;
        List<Filter> list = null;
        if (group.equals(Constants.PROVIDER)) {
            list = providerFilters;
        } else {
            list = consumerFilters;
        }
        if (list.size() > 0) {
            for (int i = list.size() - 1; i >= 0; i--) {
                final Filter filter = list.get(i);

                //如果没有启用此filter,则不加入此filter到chain
                if (excludeFilters.contains(filter.name())) {
                    continue;
                }

                final Invoker<T> next = last;
                last = new Invoker<T>() {

                    @Override
                    public Class<T> getInterface() {
                        return invoker.getInterface();
                    }

                    @Override
                    public URL getUrl() {
                        return invoker.getUrl();
                    }

                    @Override
                    public void setUrl(URL url) {
                        invoker.setUrl(url);
                    }

                    @Override
                    public boolean isAvailable() {
                        return invoker.isAvailable();
                    }

                    @Override
                    public Result invoke(RpcInvocation invocation) throws RpcException {
                        return filter.invoke(next, invocation);
                    }

                    @Override
                    public void setMethod(Method method) {
                        invoker.setMethod(method);
                    }

                    @Override
                    public Method getMethod() {
                        return invoker.getMethod();
                    }

                    @Override
                    public void setInterface(Class clazz) {
                        invoker.setInterface(clazz);
                    }

                    @Override
                    public void destroy() {
                        invoker.destroy();
                    }

                    @Override
                    public String toString() {
                        return invoker.toString();
                    }
                };
            }
        }
        return last;
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/DecoderState.java<|end_filename|>
package com.colobu.rpcx.netty;

/**
 * @author <EMAIL>
 */
public enum DecoderState {
    /**
     * 魔术值
     */
    MagicNumber,
    /**
     * 包头
     */
    Header,
    /**
     * 包体
     */
    Body,
}


<|start_filename|>demo/api/src/main/java/com/colobu/rpcx/service/DemoService.java<|end_filename|>
package com.colobu.rpcx.service;


import com.colobu.rpcx.rpc.annotation.Consumer;

/**
 * @author <EMAIL>
 */
@Consumer(impl = "com.colobu.rpcx.service.DemoServiceImpl")
public interface DemoService {

    String test(String str);
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/CompressType.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum CompressType {
    /**
     * 不压缩
     */
    None(0),
    /**
     * gzip压缩
     */
    Gzip(1);


    private final int v;

    CompressType(int v) {
        this.v = v;
    }

    private static CompressType[] values = CompressType.values();
    public static CompressType getValue(int i) {
        return values[i];
    }

    public int value() {
        return v;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyServerConfig.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.common.Config;

/**
 * @author <EMAIL>
 */
public class NettyServerConfig implements Cloneable {
    private int listenPort = 8888;
    private int serverWorkerThreads = Runtime.getRuntime().availableProcessors() * 2;
    private int serverCallbackExecutorThreads = 0;
    private int serverOnewaySemaphoreValue = 256;
    private int serverAsyncSemaphoreValue = 64;
    private int serverChannelMaxIdleTimeSeconds = 120;

    private int serverSocketSndBufSize = 1024 * 64;
    private int serverSocketRcvBufSize = 1024 * 64;

    //业务的线程数
    private int serverBizThreads = 1000;


    public int getServerBizThreads() {
        int num = Integer.valueOf(Config.ins().get("rpcx.biz.processor.num", String.valueOf(serverBizThreads)));
        return num;
    }

    public int getListenPort() {
        return listenPort;
    }


    public void setListenPort(int listenPort) {
        this.listenPort = listenPort;
    }


    public int getServerWorkerThreads() {
        int num = Integer.valueOf(Config.ins().get("rpcx.worker.processor.num", String.valueOf(serverWorkerThreads)));
        return num;
    }


    public void setServerWorkerThreads(int serverWorkerThreads) {
        this.serverWorkerThreads = serverWorkerThreads;
    }


    public int getServerOnewaySemaphoreValue() {
        return serverOnewaySemaphoreValue;
    }


    public void setServerOnewaySemaphoreValue(int serverOnewaySemaphoreValue) {
        this.serverOnewaySemaphoreValue = serverOnewaySemaphoreValue;
    }


    public int getServerCallbackExecutorThreads() {
        return serverCallbackExecutorThreads;
    }


    public void setServerCallbackExecutorThreads(int serverCallbackExecutorThreads) {
        this.serverCallbackExecutorThreads = serverCallbackExecutorThreads;
    }


    public int getServerAsyncSemaphoreValue() {
        return serverAsyncSemaphoreValue;
    }


    public void setServerAsyncSemaphoreValue(int serverAsyncSemaphoreValue) {
        this.serverAsyncSemaphoreValue = serverAsyncSemaphoreValue;
    }


    public int getServerChannelMaxIdleTimeSeconds() {
        return serverChannelMaxIdleTimeSeconds;
    }


    public void setServerChannelMaxIdleTimeSeconds(int serverChannelMaxIdleTimeSeconds) {
        this.serverChannelMaxIdleTimeSeconds = serverChannelMaxIdleTimeSeconds;
    }


    public int getServerSocketSndBufSize() {
        int size = Integer.valueOf(Config.ins().get("rpcx.snd.buf.size", String.valueOf(serverSocketSndBufSize)));
        return size;
    }


    public int getServerSocketRcvBufSize() {
        int size = Integer.valueOf(Config.ins().get("rpcx.rcv.buf.size", String.valueOf(serverSocketRcvBufSize)));
        return size;
    }


    @Override
    public Object clone() throws CloneNotSupportedException {
        return (NettyServerConfig) super.clone();
    }

}


<|start_filename|>server/src/test/java/com/colobu/rpcx/server/NettyServerTest.java<|end_filename|>
package com.colobu.rpcx.server;

import org.junit.Test;

import java.util.concurrent.TimeUnit;


/**
 * Created by <EMAIL>.
 */
public class NettyServerTest {


    @Test
    public void testNettyServer() throws InterruptedException {
        NettyServer server = new NettyServer();
        server.start();
        ZkServiceRegister reg = new ZkServiceRegister("/youpin/services/", server.getAddr() + ":" + server.getPort(), "", null);
        reg.register();
        reg.start();
        TimeUnit.HOURS.sleep(1);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/register/IServiceRegister.java<|end_filename|>
package com.colobu.rpcx.register;

/**
 * @author <EMAIL>
 */
public interface IServiceRegister {

    /**
     * 服务注册
     */
    void register();

    /**
     * 开启注册服务
     */
    void start();

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyEventExecuter.java<|end_filename|>
package com.colobu.rpcx.netty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author <EMAIL>
 */
public class NettyEventExecuter implements Runnable {


    private static final Logger logger = LoggerFactory.getLogger(NettyEventExecuter.class);

    private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();

    private final int maxSize = 10000;

    private volatile boolean stop = false;

    public void putNettyEvent(final NettyEvent event) {
        if (this.eventQueue.size() <= maxSize) {
            this.eventQueue.add(event);
        } else {
            logger.warn("event queue size[{}] enough, so drop this event {}", this.eventQueue.size(), event.toString());
        }
    }


    @Override
    public void run() {
        final ChannelEventListener listener = null;
        while (!this.isStoped()) {
            try {
                NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                if (event != null && listener != null) {
                    switch (event.getType()) {
                        case IDLE:
                            listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                            break;
                        case CLOSE:
                            listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                            break;
                        case CONNECT:
                            listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
                            break;
                        case EXCEPTION:
                            listener.onChannelException(event.getRemoteAddr(), event.getChannel());
                            break;
                        default:
                            break;

                    }
                }
            } catch (Exception e) {
                logger.warn(this.getServiceName() + " service has exception. ", e);
            }
        }
        logger.info(this.getServiceName() + " service end");
    }

    private boolean isStoped() {
        return stop;
    }

    public void shutdown() {
        this.stop = true;
    }

    private String getServiceName() {
        return null;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/selector/SelectMode.java<|end_filename|>
package com.colobu.rpcx.selector;

public enum SelectMode {
    /**
     * RandomSelect is selecting randomly
     */
    RandomSelect,
    /**
     * RoundRobin is selecting by round robin
     */
    RoundRobin,
    /**
     * WeightedRoundRobin is selecting by weighted round robin
     */
    WeightedRoundRobin,
    /**
     * SelectByUser is selecting by implementation of users
     */
    SelectByUser,
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/IClient.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.protocol.Message;

/**
 * @author <EMAIL>
 */
public interface IClient {

    Message call(Message req, long timeOut, String sendType) throws Exception;

    Message call(String addr, Message req, long timeOut) throws Exception;

    IServiceDiscovery getServiceDiscovery();

    void close();

}




<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/RpcFilterFinder.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.rpc.annotation.RpcFilter;
import org.reflections.Reflections;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Created by <EMAIL>.
 */
public class RpcFilterFinder {

    private final String filterPackage;
    private final String group;

    public RpcFilterFinder(String providerPackage, String group) {
        this.filterPackage = providerPackage;
        this.group = group;
    }


    public List<Class> find() {
        Reflections reflections = new Reflections(filterPackage);
        Set<Class<?>> classesList = reflections.getTypesAnnotatedWith(RpcFilter.class);
        List<Class> list = classesList.stream().filter(it -> {
            RpcFilter an = it.getAnnotation(RpcFilter.class);
            return Stream.of(an.group()).anyMatch(it2 -> it2.equals(group));//过滤出provider
        }).collect(Collectors.toList());
        return list.stream().sorted((a, b) -> {
            RpcFilter a1 = (RpcFilter) a.getAnnotation(RpcFilter.class);
            RpcFilter b1 = (RpcFilter) b.getAnnotation(RpcFilter.class);
            return a1.order() - b1.order();
        }).collect(Collectors.toList());
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/ConsumerConfig.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.fail.FailType;
import com.colobu.rpcx.filter.FilterWrapper;
import com.colobu.rpcx.netty.IClient;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.rpc.*;
import com.colobu.rpcx.rpc.annotation.Consumer;
import com.colobu.rpcx.selector.SelectMode;
import com.google.common.collect.Sets;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;


/**
 * @author <EMAIL>
 */
public class ConsumerConfig {

    private IClient client;

    private long timeOut = TimeUnit.SECONDS.toMillis(2);

    private int retryNum = 1;

    private String token = "";

    private String sendType = Constants.SYNC_KEY;

    private FailType failType = FailType.FailFast;

    private SelectMode selectMode = SelectMode.RandomSelect;

    private String group = "";

    public ConsumerConfig() {
    }

    public ConsumerConfig(IClient client) {
        this.client = client;
    }

    public static class ConsumerConfigBuilder {

        private ConsumerConfig config = new ConsumerConfig();

        public ConsumerConfigBuilder setClient(IClient client) {
            this.config.client = client;
            return this;
        }

        public ConsumerConfigBuilder setTimeout(long timeOut) {
            this.config.timeOut = timeOut;
            return this;
        }

        public ConsumerConfigBuilder setRetryNum(int retryNum) {
            this.config.retryNum = retryNum;
            return this;
        }

        public ConsumerConfigBuilder setToken(String token) {
            this.config.token = token;
            return this;
        }

        public ConsumerConfigBuilder setSendType(String sendType) {
            this.config.sendType = sendType;
            return this;
        }

        public ConsumerConfigBuilder setFailType(FailType failType) {
            this.config.failType = failType;
            return this;
        }

        public ConsumerConfigBuilder setSelectMode(SelectMode selectMode) {
            this.config.selectMode = selectMode;
            return this;
        }

        public ConsumerConfigBuilder setGroup(String group) {
            this.config.group = group;
            return this;
        }


        public ConsumerConfig build() {
            return this.config;
        }
    }


    public <T> T refer(Class<T> clazz) {
        return new CglibProxy().getProxy(clazz, (method, args) -> {
            Consumer provider = clazz.getAnnotation(Consumer.class);

            RpcInvocation invocation = new RpcInvocation();
            invocation.setClassName(provider.impl());
            invocation.setArguments(args);
            invocation.setMethodName(method.getName());
            invocation.setTimeOut(timeOut);
            invocation.setRetryNum(retryNum);
            invocation.setSendType(sendType);
            invocation.setFailType(failType);
            invocation.setSelectMode(selectMode);
            invocation.setLanguageCode(LanguageCode.JAVA);
            invocation.setGroup(group);

            Map<String, String> attachments = new HashMap<>(3);
            attachments.put(Constants.TOKEN_KEY, token);
            attachments.put(Constants.TRACE_ID, RpcContext.getContext().getAttachment(Constants.TRACE_ID));
            attachments.put(Constants.SPAN_ID, RpcContext.getContext().getAttachment(Constants.SPAN_ID));
            invocation.setAttachments(attachments);

            Class<?>[] types = method.getParameterTypes();
            invocation.parameterTypeNames = Stream.of(types).map(it -> ReflectUtils.getName(it)).toArray(String[]::new);
            invocation.setParameterTypes(types);
            invocation.setResultType(method.getReturnType());

            Invoker invoker = new RpcConsumerInvoker(client);
            Invoker<Object> wrapperInvoker = FilterWrapper.ins().buildInvokerChain(invoker, "", Constants.CONSUMER, Sets.newHashSet());

            Result result = wrapperInvoker.invoke(invocation);

            if (result.hasException()) {
                if (result.getException() instanceof RpcException) {
                    throw (RpcException) result.getException();
                } else {
                    throw new RpcException(result.getException());
                }
            }

            return result.getValue();
        });
    }


}


<|start_filename|>demo/demmoserver/src/main/java/com/demo/filter/DemoFilter.java<|end_filename|>
package com.demo.filter;

import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author <EMAIL>
 */
@RpcFilter(group = {Constants.PROVIDER})
public class DemoFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(DemoFilter.class);

    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        logger.info("----------->demoFilter url:{}", invocation.url.toFullString());
        return invoker.invoke(invocation);
    }
}


<|start_filename|>client/src/test/java/com/colobu/rpcx/client/SelectorTest.java<|end_filename|>
package com.colobu.rpcx.client;

import com.colobu.rpcx.client.impl.RandomSelector;
import org.junit.Test;

/**
 * Created by zhangzhiyong on 2018/7/4.
 */
public class SelectorTest {


    @Test
    public void testRandomSelector() {
        System.out.println(new RandomSelector().select("","", null));
    }
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/rpc/ProxyTest.java<|end_filename|>
package com.colobu.rpcx.rpc;

import org.junit.Test;

/**
 * Created by <EMAIL>.
 */
public class ProxyTest {



    @Test
    public void testCglib() {
        IA proxy = new CglibProxy().getProxy(IA.class,(m,args)->{
            return "dddd";
        });
        String res = proxy.call();
        System.out.println(res);

    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/HessianUtils.java<|end_filename|>
package com.colobu.rpcx.rpc;

import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;

/**
 * @author <EMAIL>
 */
public class HessianUtils {


    private HessianUtils() {

    }

    public static byte[] write(Object obj) {
        ByteArrayOutputStream os = new ByteArrayOutputStream();
        HessianOutput ho = new HessianOutput(os);
        try {
            ho.writeObject(obj);
            return os.toByteArray();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                ho.close();
                os.close();
            } catch (Exception ex) {

            }
        }
    }


    public static Object read(byte[] data) {
        ByteArrayInputStream is = new ByteArrayInputStream(data);
        HessianInput hi = new HessianInput(is);
        try {
            return hi.readObject();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            try {
                hi.close();
                is.close();
            } catch (Exception ex) {

            }
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/ChannelWrapper.java<|end_filename|>
package com.colobu.rpcx.netty;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

/**
 * @author <EMAIL>
 */
public class ChannelWrapper {

    private final ChannelFuture channelFuture;


    public ChannelWrapper(ChannelFuture channelFuture) {
        this.channelFuture = channelFuture;
    }


    public boolean isOK() {
        return this.channelFuture.channel() != null && this.channelFuture.channel().isActive();
    }


    public boolean isWriteable() {
        return this.channelFuture.channel().isWritable();
    }


    public Channel getChannel() {
        return this.channelFuture.channel();
    }


    public ChannelFuture getChannelFuture() {
        return channelFuture;
    }

}


<|start_filename|>server/src/main/java/com/colobu/rpcx/handler/RpcxHttpHandler.java<|end_filename|>
package com.colobu.rpcx.handler;

import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.protocol.LanguageCode;
import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.MessageType;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.server.NettyServer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.socks.SocksMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.HashMap;

/**
 * XVersion           = "X-RPCX-Version"
 * XMessageType       = "X-RPCX-MesssageType"
 * XHeartbeat         = "X-RPCX-Heartbeat"
 * XOneway            = "X-RPCX-Oneway"
 * XMessageStatusType = "X-RPCX-MessageStatusType"
 * XSerializeType     = "X-RPCX-SerializeType"
 * XMessageID         = "X-RPCX-MessageID"
 * XServicePath       = "X-RPCX-ServicePath"
 * XServiceMethod     = "X-RPCX-ServiceMethod"
 * XMeta              = "X-RPCX-Meta"
 * XErrorMessage      = "X-RPCX-ErrorMessage"
 *
 * @author <EMAIL>
 */
public class RpcxHttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private static final Logger logger = LoggerFactory.getLogger(RpcxHttpHandler.class);

    private NettyServer nettyServer;

    public RpcxHttpHandler(NettyServer nettyServer) {
        this.nettyServer = nettyServer;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) {
        String servicePath = msg.headers().get(Constants.X_RPCX_SERVICEPATH);
        String serviceMethod = msg.headers().get(Constants.X_RPCX_SERVICEMETHOD);
        String traceId = msg.headers().get(Constants.X_RPCX_TRACEID);

        logger.info("service:{} method:{} traceId:{}", servicePath, serviceMethod, traceId);

        ByteBuf buf = msg.content();
        int len = buf.readableBytes();

        byte[] payload = new byte[len];
        buf.readBytes(payload);

        Message message = new Message();
        message.metadata.put("language", LanguageCode.HTTP.name());
        message.metadata.put("_host", getClientIp(ctx, msg));
        message.metadata.put("_port", "0");
        message.metadata.put(Constants.X_RPCX_TRACEID, traceId == null ? "" : traceId);
        message.servicePath = servicePath;
        message.serviceMethod = serviceMethod;
        message.payload = payload;
        message.setMessageType(MessageType.Request);
        message.setOneway(true);
        RemotingCommand command = RemotingCommand.createRequestCommand(message);
        command.setCode(1984);
        //这里会异步处理
        nettyServer.processRequestCommand(ctx, command);
    }


    private String getClientIp(ChannelHandlerContext ctx, FullHttpRequest request) {
        HttpHeaders httpHeaders = request.headers();
        String remoteIP = httpHeaders.get("X-Forwarded-For");
        if (remoteIP == null) {
            remoteIP = httpHeaders.get("X-Real-IP");
        }
        if (remoteIP != null) {
            return remoteIP;
        } else { // request取不到就从channel里取
            return ((InetSocketAddress) ctx.channel().remoteAddress()).getHostName();
        }
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/cache/AbstractCacheFactory.java<|end_filename|>
package com.colobu.rpcx.cache;


import com.colobu.rpcx.rpc.URL;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author <EMAIL>
 */
public abstract class AbstractCacheFactory implements CacheFactory {
    
    private final ConcurrentMap<String, Cache> caches = new ConcurrentHashMap<String, Cache>();

    @Override
    public Cache getCache(URL url) {
        String key = url.toFullString();
        Cache cache = caches.get(key);
        if (cache == null) {
            caches.put(key, createCache(url));
            cache = caches.get(key);
        }
        return cache;
    }

    protected abstract Cache createCache(URL url);

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/ReflectUtils.java<|end_filename|>
package com.colobu.rpcx.rpc;

import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


/**
 * Created by <EMAIL>.
 */
public class ReflectUtils {

    private static final ConcurrentMap<String, Class<?>> NAME_CLASS_CACHE = new ConcurrentHashMap<String, Class<?>>();


    /**
     * void(V).
     */
    public static final char JVM_VOID = 'V';

    /**
     * boolean(Z).
     */
    public static final char JVM_BOOLEAN = 'Z';

    /**
     * byte(B).
     */
    public static final char JVM_BYTE = 'B';

    /**
     * char(C).
     */
    public static final char JVM_CHAR = 'C';

    /**
     * double(D).
     */
    public static final char JVM_DOUBLE = 'D';

    /**
     * float(F).
     */
    public static final char JVM_FLOAT = 'F';

    /**
     * int(I).
     */
    public static final char JVM_INT = 'I';

    /**
     * long(J).
     */
    public static final char JVM_LONG = 'J';

    /**
     * short(S).
     */
    public static final char JVM_SHORT = 'S';


    public static String getName(Class<?> c) {
        String desc = getDesc(c);
        return desc2name(desc);
    }

    public static String getDesc(Class<?> c) {
        StringBuilder ret = new StringBuilder();

        while (c.isArray()) {
            ret.append('[');
            c = c.getComponentType();
        }

        if (c.isPrimitive()) {
            String t = c.getName();
            if ("void".equals(t)) {
                ret.append(JVM_VOID);
            } else if ("boolean".equals(t)) {
                ret.append(JVM_BOOLEAN);
            } else if ("byte".equals(t)) {
                ret.append(JVM_BYTE);
            } else if ("char".equals(t)) {
                ret.append(JVM_CHAR);
            } else if ("double".equals(t)) {
                ret.append(JVM_DOUBLE);
            } else if ("float".equals(t)) {
                ret.append(JVM_FLOAT);
            } else if ("int".equals(t)) {
                ret.append(JVM_INT);
            } else if ("long".equals(t)) {
                ret.append(JVM_LONG);
            } else if ("short".equals(t)) {
                ret.append(JVM_SHORT);
            }
        } else {
            ret.append('L');
            ret.append(c.getName().replace('.', '/'));
            ret.append(';');
        }
        return ret.toString();
    }


    public static String desc2name(String desc) {
        StringBuilder sb = new StringBuilder();
        int c = desc.lastIndexOf('[') + 1;
        if (desc.length() == c + 1) {
            switch (desc.charAt(c)) {
                case JVM_VOID: {
                    sb.append("void");
                    break;
                }
                case JVM_BOOLEAN: {
                    sb.append("boolean");
                    break;
                }
                case JVM_BYTE: {
                    sb.append("byte");
                    break;
                }
                case JVM_CHAR: {
                    sb.append("char");
                    break;
                }
                case JVM_DOUBLE: {
                    sb.append("double");
                    break;
                }
                case JVM_FLOAT: {
                    sb.append("float");
                    break;
                }
                case JVM_INT: {
                    sb.append("int");
                    break;
                }
                case JVM_LONG: {
                    sb.append("long");
                    break;
                }
                case JVM_SHORT: {
                    sb.append("short");
                    break;
                }
                default:
                    throw new RuntimeException();
            }
        } else {
            sb.append(desc.substring(c + 1, desc.length() - 1).replace('/', '.'));
        }
        while (c-- > 0) {
            sb.append("[]");
        }
        return sb.toString();
    }


    public static Class<?> forName(String name) {
        try {
            return name2class(name);
        } catch (ClassNotFoundException e) {
            throw new IllegalStateException("Not found class " + name + ", cause: " + e.getMessage(), e);
        }
    }


    public static Class<?> name2class(String name) throws ClassNotFoundException {
        return name2class(Thread.currentThread().getClass().getClassLoader(), name);
    }

    private static Class<?> name2class(ClassLoader cl, String name) throws ClassNotFoundException {
        int c = 0, index = name.indexOf('[');
        if (index > 0) {
            c = (name.length() - index) / 2;
            name = name.substring(0, index);
        }
        if (c > 0) {
            StringBuilder sb = new StringBuilder();
            while (c-- > 0) {
                sb.append("[");
            }

            if ("void".equals(name)) {
                sb.append(JVM_VOID);
            } else if ("boolean".equals(name)) {
                sb.append(JVM_BOOLEAN);
            } else if ("byte".equals(name)) {
                sb.append(JVM_BYTE);
            } else if ("char".equals(name)) {
                sb.append(JVM_CHAR);
            } else if ("double".equals(name)) {
                sb.append(JVM_DOUBLE);
            } else if ("float".equals(name)) {
                sb.append(JVM_FLOAT);
            } else if ("int".equals(name)) {
                sb.append(JVM_INT);
            } else if ("long".equals(name)) {
                sb.append(JVM_LONG);
            } else if ("short".equals(name)) {
                sb.append(JVM_SHORT);
            } else {
                sb.append('L').append(name).append(';'); // "java.lang.Object" ==> "Ljava.lang.Object;"
            }
            name = sb.toString();
        } else {
            if ("void".equals(name)) {
                return void.class;
            } else if ("boolean".equals(name)) {
                return boolean.class;
            } else if ("byte".equals(name)) {
                return byte.class;
            } else if ("char".equals(name)) {
                return char.class;
            } else if ("double".equals(name)) {
                return double.class;
            } else if ("float".equals(name)) {
                return float.class;
            } else if ("int".equals(name)) {
                return int.class;
            } else if ("long".equals(name)) {
                return long.class;
            } else if ("short".equals(name)) {
                return short.class;
            }
        }

        if (cl == null) {
            cl = ClassHelper.getClassLoader();
        }
        Class<?> clazz = NAME_CLASS_CACHE.get(name);
        if (clazz == null) {
            clazz = Class.forName(name, true, cl);
            NAME_CLASS_CACHE.put(name, clazz);
        }
        return clazz;
    }


    public static boolean isPrimitives(Class<?> cls) {
        if (cls.isArray()) {
            return isPrimitive(cls.getComponentType());
        }
        return isPrimitive(cls);
    }


    public static boolean isPrimitive(Class<?> cls) {
        return cls.isPrimitive() || cls == String.class || cls == Boolean.class || cls == Character.class
                || Number.class.isAssignableFrom(cls) || Date.class.isAssignableFrom(cls);
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/RpcxVersion.java<|end_filename|>
package com.colobu.rpcx.common;

/**
 * @author <EMAIL>
 */
public class RpcxVersion {

    private String version = "1.2.6";
    private String date = "20181102";


    @Override
    public String toString() {
        return version + " " + date;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/ProviderFinder.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.rpc.annotation.Provider;
import org.reflections.Reflections;

import java.util.Set;


/**
 * @author <EMAIL>
 */
public class ProviderFinder {

    private final String providerPackage;

    public ProviderFinder(String providerPackage) {
        this.providerPackage = providerPackage;
    }


    public Set<Class<?>> find() {
        Reflections reflections = new Reflections(providerPackage);
        Set<Class<?>> classesList = reflections.getTypesAnnotatedWith(Provider.class);
        return classesList;
    }

}


<|start_filename|>demo/democlient/src/main/resources/client/nodejs/client.js<|end_filename|>
var request = require('request');

var options = {
    url: 'http://10.231.72.221:8033/',
    headers: {
        'User-Agent': 'Easy_Go_NodeJs_Client',
        'X-RPCX-ServicePath':'com.colobu.rpcx.service.TestService',
        'X-RPCX-ServiceMethod':'sum',
        'connection':'close'
    },
    method: 'POST',
    json:true,
    body:{"parameterTypeNames":["int","int"],"arguments":["11","22"]}
};

function callback(error, response, body) {
    if (!error && response.statusCode == 200) {
        console.log(body);
        console.log(body.payload)
        console.log(new Buffer(body.payload).toString('ascii'))
    } else {
        console.log(error)
    }
}

request(options, callback)

<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/SerializeType.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum SerializeType {

    /**
     * java序列化
     */
    SerializeNone(0),
    /**
     * json序列化
     */
    JSON(1),
    /**
     * protobuf序列化
     */
    ProtoBuffer(2),
    /**
     * msgpack序列化
     */
    MsgPack(3),


    Thrift(4);


    private final int v;

    SerializeType(int v) {
        this.v = v;
    }

    private static SerializeType[] values = SerializeType.values();
    public static SerializeType getValue(int i) {
        return values[i];
    }

    public int value() {
        return v;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/discovery/IServiceDiscovery.java<|end_filename|>
package com.colobu.rpcx.discovery;

import java.util.List;

/**
 * @author <EMAIL>
 */
public interface IServiceDiscovery {

    /**
     * 根据serviceName 获取 服务的ip和端口号列表
     * @param serviceName
     * @return
     */
    List<String> getServices(String serviceName);

    /**
     * 观察注册中心的变化
     */
    void watch();

    void close();

    /**
     * 手动添加需要监控的服务
     * @param serviceNames
     */
    void addServices(final String... serviceNames);
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/common/WeightedRountRobinTest.java<|end_filename|>
package com.colobu.rpcx.common;

import com.colobu.rpcx.selector.Weighted;
import com.colobu.rpcx.selector.WeightedRountRobin;
import com.google.common.collect.Lists;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class WeightedRountRobinTest {


    @Test
    public void testWeightedRountRobin() {
        Weighted[] array = new Weighted[3];
        array[0] = new Weighted("a", 5);
        array[1] = new Weighted("b", 1);
        array[2] = new Weighted("c", 1);

        WeightedRountRobin weightedRountRobin = new WeightedRountRobin(array);


        weightedRountRobin.updateWeighteds(Arrays.asList());

        for (int i = 0; i < 7; i++) {
            Weighted a = weightedRountRobin.nextWeighted();
            System.out.println(a.getServer());
        }
    }


    @Test
    public void testUpdateWeightedRountRobin() {

        List<String> serviceList = Arrays.asList("127.0.0.1:8001?weight=10");
        WeightedRountRobin weightedRountRobin = new WeightedRountRobin(serviceList);
        weightedRountRobin.updateWeighteds(Arrays.asList("127.0.0.1:8002?weight=10","127.0.0.1:8001?weight=9"));

    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/ClassPathResource.java<|end_filename|>
package com.colobu.rpcx.common;

import com.google.common.io.ByteStreams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

/**
 * @author <EMAIL>
 */
public class ClassPathResource {

    private static final Logger logger = LoggerFactory.getLogger(ClassPathResource.class);

    private final String path;

    private ClassLoader classLoader;


    public ClassPathResource(String path) {
        this.path = path;
        this.classLoader = ClassUtils.getDefaultClassLoader();
    }


    public InputStream getInputStream() {
        InputStream is;
        if (this.classLoader != null) {
            is = this.classLoader.getResourceAsStream(this.path);
        } else {
            is = ClassLoader.getSystemResourceAsStream(this.path);
        }
        if (is == null) {
            logger.warn("cannot be opened because it does not exist:{}", this.path);
        }
        return is;
    }


    public String getString() {
        try {
            InputStream is = getInputStream();
            if (null == is) {
                return "";
            }
            return new String(ByteStreams.toByteArray(is));
        } catch (IOException e) {
            e.printStackTrace();
            return "";
        }
    }

    public Properties getProperties() {
        Properties prop = new Properties();
        try {
            InputStream is = getInputStream();
            if (null == is) {
                return prop;
            }
            prop.load(is);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return prop;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/ContextFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcContext;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author <EMAIL>
 */
@RpcFilter(order = -2000, group = {Constants.PROVIDER})
public class ContextFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(ContextFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        logger.debug("ContextFilter begin className:{} methodName:{}", invocation.getClassName(), invocation.getMethodName());
        Map<String, String> attachments = invocation.getAttachments();
        RpcContext.getContext()
                .setAttachments(attachments)
                .setRemoteAddress(invocation.getUrl().getAddress(), invocation.getUrl().getPort())
                .setLocalAddress(invoker.getUrl().getHost(),
                        invoker.getUrl().getPort());
        invocation.setInvoker(invoker);
        try {
            return invoker.invoke(invocation);
        } finally {
            RpcContext.removeContext();
            logger.debug("ContextFilter end");
        }
    }


}

<|start_filename|>common/src/test/java/com/colobu/rpcx/rpc/IA.java<|end_filename|>
package com.colobu.rpcx.rpc;

public interface IA {

    String call();


    int sum(int a, int b);
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyDecoder.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.protocol.Message;
import com.colobu.rpcx.protocol.RemotingCommand;
import com.colobu.rpcx.rpc.RpcException;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

/**
 * @author <EMAIL>
 */
public class NettyDecoder extends ReplayingDecoder<DecoderState> {

    private Message message = null;

    public NettyDecoder() {
        super(DecoderState.MagicNumber);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        switch (state()) {
            case MagicNumber:
                message = new Message();
                byte magicNumber = in.readByte();
                if (magicNumber != Message.magicNumber) {
                    throw new RpcException("magicNumber error:" + magicNumber);
                }
                checkpoint(DecoderState.Header);
            case Header:
                byte[] header = new byte[12];
                header[0] = Message.magicNumber;
                in.readBytes(header, 1, 11);
                message.header = header;
                checkpoint(DecoderState.Body);
            case Body:
                int totalLen = in.readInt();
                byte[] data = new byte[totalLen];
                in.readBytes(data);
                RemotingCommand command = new RemotingCommand(message, data);
                checkpoint(DecoderState.MagicNumber);
                //业务解码交给业务层
                out.add(command);
                break;
            default:
                throw new RpcException("Shouldn't reach here.");
        }
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/CglibProxy.java<|end_filename|>
package com.colobu.rpcx.rpc;

import net.sf.cglib.proxy.Enhancer;

import java.lang.reflect.Method;
import java.util.function.BiFunction;

/**
 * Created by <EMAIL>.
 */
public class CglibProxy {


    public <T> T getProxy(Class<?> clazz, final BiFunction<Method, Object[], Object> function) {
        Enhancer e = new Enhancer();
        e.setSuperclass(clazz);
        e.setCallback((net.sf.cglib.proxy.InvocationHandler) (proxy, method, args) -> {
            String methodName = method.getName();
            if ("getClass".equals(methodName)) {
                return proxy.getClass();
            }
            if ("hashCode".equals(methodName)) {
                return proxy.hashCode();
            }
            if ("toString".equals(methodName)) {
                return proxy.toString();
            }
            return function.apply(method, args);
        });
        return (T) e.create();
    }


}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/config/ConsumerConfig.java<|end_filename|>
package com.colobu.rpcx.config;


import com.colobu.rpcx.rpc.impl.ConsumerConfig.ConsumerConfigBuilder;
import com.colobu.rpcx.service.ITest2Service;
import com.colobu.rpcx.spring.RpcxConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author <EMAIL>
 */
@Configuration
public class ConsumerConfig {

    @Autowired
    private RpcxConsumer rpcxConsumer;


    @Bean
    public ITest2Service test2Service() {
        return rpcxConsumer.wrap(ITest2Service.class, new ConsumerConfigBuilder().setRetryNum(3));
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/ConsumerTraceidFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author <EMAIL>
 * 打印traceid
 */
@RpcFilter(group = {Constants.CONSUMER}, order = -1000)
public class ConsumerTraceidFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerTraceidFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        Result result = invoker.invoke(invocation);

        if (null != result) {
            String key = ClassUtils.getMethodKey(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
            String traceId = result.getAttachment(Constants.TRACE_ID);
            String spanId = result.getAttachment(Constants.SPAN_ID);
            logger.info("invoke {} traceid:{} spanId:{} error:{}", key, traceId, spanId, result.hasException());
        }

        return result;
    }
}


<|start_filename|>server/src/main/java/com/colobu/rpcx/server/ZkServiceRegister.java<|end_filename|>
package com.colobu.rpcx.server;

import com.colobu.rpcx.register.IServiceRegister;
import com.colobu.rpcx.rpc.impl.Exporter;
import com.colobu.rpcx.utils.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * @author <EMAIL>
 */
public class ZkServiceRegister implements IServiceRegister {

    private static final Logger logger = LoggerFactory.getLogger(ZkServiceRegister.class);

    private String basePath;
    private Set<String> serviceNameSet;
    private String addr;

    public ZkServiceRegister(String basePath, String addr, String providerPackage, Function<Class, Object> getBeanFunc) {
        this.basePath = basePath;
        this.addr = addr;
        //导出所有有注解的service
        this.serviceNameSet = new Exporter(getBeanFunc, addr).export(providerPackage);
        logger.info("export service names:{}", this.serviceNameSet);
    }

    /**
     * 服务注册
     */
    @Override
    public void register() {
        try {
            ZkClient.ins().create(this.basePath, this.serviceNameSet, this.addr);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 周期性的注册
     */
    @Override
    public void start() {
        new ScheduledThreadPoolExecutor(1).scheduleWithFixedDelay(() -> {
            register();
        }, 0, 5, TimeUnit.SECONDS);
    }

}


<|start_filename|>demo/democlient/src/main/java/com/colobu/rpcx/demo/Client.java<|end_filename|>
package com.colobu.rpcx.demo;

import com.colobu.rpcx.discovery.IServiceDiscovery;
import com.colobu.rpcx.client.NettyClient;
import com.colobu.rpcx.client.ZkServiceDiscovery;
import com.colobu.rpcx.netty.IClient;
import com.colobu.rpcx.rpc.impl.ConsumerConfig;
import com.colobu.rpcx.service.IArith;


/**
 * @author <EMAIL>
 */
public class Client {

    public static void main(String... args) {
        IServiceDiscovery serviceDiscovery = new ZkServiceDiscovery("/youpin/services/","");
        IClient client = new NettyClient(serviceDiscovery);
        IArith arith = new ConsumerConfig(client).refer(IArith.class);
////        System.out.println(arith.sum(1111, 222));
////        System.out.println(arith.sum2(1111, 222));
        for (int i = 0; i < 1; i++) {
            System.out.println(arith.sum(11,22));
        }
//
        serviceDiscovery.close();
        System.out.println("client call finish");
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/MessageStatusType.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public enum MessageStatusType {
    /**
     * 一般
     */
    Normal(0),
    /**
     * 错误
     */
    Error(1);


    private final int v;

    MessageStatusType(int v) {
        this.v = v;
    }

    private static MessageStatusType[] values = MessageStatusType.values();
    public static MessageStatusType getValue(int i) {
        return values[i];
    }

    public int value() {
        return v;
    }
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/filter/FilterTest.java<|end_filename|>
package com.colobu.rpcx.filter;

import com.colobu.rpcx.filter.impl.EchoFilter;
import org.junit.Test;

public class FilterTest {


    @Test
    public void testName() {
        EchoFilter filter = new EchoFilter();
        System.out.println(filter.name());
    }
}


<|start_filename|>client/src/main/java/com/colobu/rpcx/client/Selector.java<|end_filename|>
package com.colobu.rpcx.client;

import java.util.List;

/**
 * Created by <EMAIL>.
 */
public interface Selector {


    String select(String servicePath, String serviceMethod, List<String> serviceList);

}


<|start_filename|>common/src/test/java/com/colobu/rpcx/protocol/MessageTest.java<|end_filename|>
package com.colobu.rpcx.protocol;

import org.junit.Test;


public class MessageTest {


    @Test
    public void encodeAndDecode() {

    }
}

<|start_filename|>demo/api/src/main/java/com/colobu/rpcx/service/IArith.java<|end_filename|>
package com.colobu.rpcx.service;

import com.colobu.rpcx.rpc.annotation.Consumer;

/**
 * @author <EMAIL>
 */
@Consumer(impl = "com.colobu.rpcx.service.Arith")
public interface IArith {

    Integer sum(Integer a, Integer b);

    int sum2(int a, int b);

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/RpcException.java<|end_filename|>
package com.colobu.rpcx.rpc;


/**
 * @author <EMAIL>
 */
public class RpcException extends RuntimeException {

    private String code;

    public RpcException(String message, String code) {
        super(message);
        this.code = code;
    }

    public RpcException(String message, Throwable cause, String code) {
        super(message, cause);
        this.code = code;
    }

    public RpcException(Throwable cause) {
        super(cause);
    }

    public RpcException(String message) {
        super(message);
    }

    public RpcException(Exception ex) {
        super(ex);
    }


    public String getCode() {
        return code;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/retry/RetryPolicy.java<|end_filename|>
package com.colobu.rpcx.common.retry;


import com.colobu.rpcx.rpc.Result;

import java.util.function.Function;

/**
 * @author <EMAIL>
 */
public interface RetryPolicy {

    /**
     * 判断的条件是result中没有异常
     * @param func
     * @return
     */
    Result retry(Function<Integer, Result> func);
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/rpc/impl/RpcResult.java<|end_filename|>
package com.colobu.rpcx.rpc.impl;

import com.colobu.rpcx.rpc.Result;
import java.util.HashMap;
import java.util.Map;

/**
 * @author <EMAIL>
 */
public class RpcResult implements Result {

    private Object value;

    private Throwable throwable;

    private Map<String, String> attachments = new HashMap<>();

    public RpcResult(Object value) {
        this.value = value;
    }

    public RpcResult() {
    }


    public RpcResult(Throwable throwable) {
        this.throwable = throwable;
    }

    @Override
    public void setValue(Object value) {
        this.value = value;
    }

    @Override
    public Object getValue() {
        return value;
    }

    @Override
    public Throwable getException() {
        return this.throwable;
    }

    @Override
    public void setThrowable(Throwable throwable) {
        this.throwable = throwable;
    }

    @Override
    public boolean hasException() {
        return null != throwable;
    }

    @Override
    public Map<String, String> getAttachments() {
        return attachments;
    }

    @Override
    public String getAttachment(String key) {
        return this.attachments.get(key);
    }

    @Override
    public String getAttachment(String key, String defaultValue) {
        String result = attachments.get(key);
        if (result == null || result.length() == 0) {
            result = defaultValue;
        }
        return result;
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/TimeoutFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;


/**
 * @author <EMAIL>
 */
@RpcFilter(order = -1000, group = {Constants.PROVIDER})
public class TimeoutFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(TimeoutFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        if (!("-1".equals(invoker.getUrl().getParameter(Constants.TIMEOUT_KEY)))) {
            long start = System.currentTimeMillis();
            Result result = invoker.invoke(invocation);
            long elapsed = System.currentTimeMillis() - start;

            int v = invoker.getUrl().getParameter(Constants.TIMEOUT_KEY, Integer.MAX_VALUE);
            if (invoker.getUrl() != null && v != 0 && elapsed > v) {
                logger.warn("invoke time out. method: " + invocation.getMethodName()
                        + " arguments: " + Arrays.toString(invocation.getArguments()) + " , url is "
                        + invoker.getUrl() + ", invoke elapsed " + elapsed + " ms.");
            }
            return result;
        }

        return invoker.invoke(invocation);
    }

}

<|start_filename|>common/src/main/java/com/colobu/rpcx/common/retry/RetryNTimes.java<|end_filename|>
package com.colobu.rpcx.common.retry;

import com.colobu.rpcx.rpc.Result;

import java.util.function.Function;

/**
 * @author <EMAIL>
 */
public class RetryNTimes implements RetryPolicy {


    private final int n;

    public RetryNTimes(int n) {
        this.n = n;
    }

    @Override
    public Result retry(Function<Integer, Result> func) {
        Result res = null;
        for (int i = 1; i < n + 1; i++) {
            res = func.apply(i);
            if (!res.getAttachment("needRetry").equals("true")) {
                return res;
            }
        }
        return res;
    }
}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/service/TestService.java<|end_filename|>
package com.colobu.rpcx.service;

import com.colobu.rpcx.rpc.annotation.Provider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Random;
import java.util.concurrent.TimeUnit;


/**
 * @author <EMAIL>
 */
@Service
//@Provider(name = "TestService", version = "0.0.2", token = "zz<PASSWORD>", timeout = 1000, cache = true)
//@Provider(name = "TestService", version = "0.0.2", token = "zz<PASSWORD>", tps = 1)
//@Provider(name = "TestService", version = "0.0.2", weight = "service_weight", group = "test")
@Provider(name = "TestService", version = "0.0.2", weight = "rpc.service.weight")
public class TestService implements ITestService {

    private static final Logger logger = LoggerFactory.getLogger(TestService.class);


    @Autowired
    private ITest2Service test2Service;

    private Random random = new Random();

    public String hi(String str) {

//        logger.info("-------------->call hi:{}", str);
//        int i = random.nextInt(6);
//        if (i != 5) {
//            logger.info("-------------->|||call error " + i);
//            throw new RuntimeException("call hi error!!!!");
//        }
//        logger.info("-------------->call success " + i);

//        try {
//            TimeUnit.SECONDS.sleep(15);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        return "hi ####" + str;
    }

    public String async() {
        return "result:" + System.currentTimeMillis();
    }


    public byte[] golangHi(byte[] data) {
        return ("hi " + new String(data)).getBytes();
    }

    public int sum(int a, int b) {
        return a + b;
    }

    /**
     * 再次调用其他服务
     * @param a
     * @param b
     * @return
     */
    public int sum2(int a, int b) {
        return a + b + test2Service.num();
    }

    /**
     * 测试错误信息
     *
     * @param a
     * @param b
     * @return
     */
    public int error(int a, int b) {
        if (1 == 1) {
            throw new RuntimeException("system error!!!!");
        }
        return a + b;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/tps/DefaultTPSLimiter.java<|end_filename|>
package com.colobu.rpcx.tps;


import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.rpc.Invocation;
import com.colobu.rpcx.rpc.URL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author <EMAIL>
 */
public class DefaultTPSLimiter implements TPSLimiter {

    private static final Logger logger = LoggerFactory.getLogger(DefaultTPSLimiter.class);

    private final ConcurrentMap<String, StatItem> stats = new ConcurrentHashMap<>();

    @Override
    public boolean isAllowable(URL url, Invocation invocation) {
        int rate = url.getParameter(Constants.TPS_LIMIT_RATE_KEY, -1);
        //默认时间间隔是60秒
        long interval = url.getParameter(Constants.TPS_LIMIT_INTERVAL_KEY, Constants.DEFAULT_TPS_LIMIT_INTERVAL);
        String serviceKey = url.getServiceKey();
        logger.debug("DefaultTPSLimiter interval:{} serviceKey:{}",interval,serviceKey);
        if (rate > 0) {
            StatItem statItem = stats.get(serviceKey);
            if (statItem == null) {
                stats.putIfAbsent(serviceKey,
                        new StatItem(serviceKey, rate, interval));
                statItem = stats.get(serviceKey);
            }
            return statItem.isAllowable();
        } else {//如果不开启,删除之前记录的即可
            StatItem statItem = stats.get(serviceKey);
            if (statItem != null) {
                stats.remove(serviceKey);
            }
        }

        return true;
    }

}


<|start_filename|>common/src/main/java/com/colobu/rpcx/protocol/CommandCustomHeader.java<|end_filename|>
package com.colobu.rpcx.protocol;

/**
 * Created by <EMAIL>.
 */
public class CommandCustomHeader {
}


<|start_filename|>common/src/test/java/com/colobu/rpcx/rpc/ReflectUtilsTest.java<|end_filename|>
package com.colobu.rpcx.rpc;

import org.junit.Test;

import java.lang.reflect.Method;
import java.util.Arrays;

/**
 * Created by <EMAIL>.
 */
public class ReflectUtilsTest {


    @Test
    public void testForName() throws ClassNotFoundException {
        int i = 12;
        System.out.println(ReflectUtils.getDesc(int.class));
        System.out.println(ReflectUtils.getDesc(A.class));


        String str = ReflectUtils.getDesc(int.class);
        System.out.println(ReflectUtils.name2class(ReflectUtils.desc2name(str)));
    }


    @Test
    public void testMethod() throws NoSuchMethodException {
        Method m = IA.class.getMethod("sum", int.class, int.class);
        System.out.println(m);

        Class<?>[] types = m.getParameterTypes();
        System.out.println(Arrays.toString(types));

        Class<?> resType = m.getReturnType();
        System.out.println(resType);
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/tps/TPSLimiter.java<|end_filename|>
package com.colobu.rpcx.tps;


import com.colobu.rpcx.rpc.Invocation;
import com.colobu.rpcx.rpc.URL;

public interface TPSLimiter {

    /**
     * 根据 tps 限流规则判断是否限制此次调用.
     *
     * @param url url
     * @param invocation invocation
     * @return true 则允许调用，否则不允许
     */
    boolean isAllowable(URL url, Invocation invocation);

}


<|start_filename|>server/src/main/java/com/colobu/rpcx/server/NettyServer.java<|end_filename|>
package com.colobu.rpcx.server;

import com.colobu.rpcx.common.*;
import com.colobu.rpcx.handler.RpcxProcessHandler;
import com.colobu.rpcx.netty.*;
import com.colobu.rpcx.processor.RpcHttpProcessor;
import com.colobu.rpcx.processor.RpcProcessor;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.*;
import java.util.function.Function;


/**
 * @author <EMAIL>
 */
public class NettyServer extends NettyRemotingAbstract {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private final ServerBootstrap serverBootstrap;

    private String addr;

    private int port;

    private final CountDownLatch latch = new CountDownLatch(1);

    private final ScheduledThreadPoolExecutor timer = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("timer"));

    private final ScheduledThreadPoolExecutor processingRequestSchedule = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("processingRequestSchedule"));

    private final EventLoopGroup eventLoopGroupBoss;
    private final EventLoopGroup eventLoopGroupWorker;
    private final NettyServerConfig nettyServerConfig;


    private boolean rpcxDebug = Boolean.valueOf(Config.ins().get("rpcx.debug", "false"));

    /**
     * 如果使用ioc容器,这里需要注入获取bean的 function
     */
    private Function<Class, Object> getBeanFunc;

    public NettyServer() {
        nettyServerConfig = new NettyServerConfig();
        this.serverBootstrap = new ServerBootstrap();


        if (useEpoll()) {
            logger.info("----->use epollEventLoopGroup");
            this.eventLoopGroupBoss = new EpollEventLoopGroup(1, new NamedThreadFactory("EpollNettyServerBoss_", false));
            this.eventLoopGroupWorker = new EpollEventLoopGroup(nettyServerConfig.getServerWorkerThreads(), new NamedThreadFactory("NettyEpollServerWorker_", false));
        } else {
            this.eventLoopGroupBoss = new NioEventLoopGroup(1, new NamedThreadFactory("NettyServerBoss_", false));
            this.eventLoopGroupWorker = new NioEventLoopGroup(nettyServerConfig.getServerWorkerThreads(), new NamedThreadFactory("NettyServerWorker_", false));
        }

    }


    public void start() {
        //默认的处理器
        this.defaultRequestProcessor = createDefaultRequestProcessor();
        //http网关请求处理器
        int httpProcessorNum = Integer.valueOf(Config.ins().get("rpcx.http.processor.num", "50"));
        this.registerProcessor(1984, new RpcHttpProcessor(this.getBeanFunc), new ThreadPoolExecutor(httpProcessorNum, httpProcessorNum,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), new NamedThreadFactory("httpRequestProcessor")));


        ServerBootstrap childHandler = createServerBootstrap();
        try {
            String host = RemotingUtil.getLocalAddress();
            String port = getServerPort();
            ChannelFuture sync = this.serverBootstrap.bind(host, Integer.parseInt(port)).sync();
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            logger.info("----->server bind finish");
            this.addr = addr.getHostString();
            this.port = addr.getPort();
            logger.info("###########rpc server start addr:{} ", this.addr + ":" + this.port);
        } catch (InterruptedException e1) {
            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
        }

        if (this.channelEventListener != null) {
            new Thread(this.nettyEventExecuter).start();
        }

        runScanResponseTableSchedule();
        runProcessingRequestSchedule();
        addShutdownHook();
    }

    private Pair<NettyRequestProcessor, ExecutorService> createDefaultRequestProcessor() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(nettyServerConfig.getServerBizThreads(), nettyServerConfig.getServerBizThreads(),
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), new NamedThreadFactory("DefaultRequestProcessorPool"));

        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            if (rpcxDebug) {
                try {
                    logger.info("default request pool----->{} {} {}", executor.getActiveCount(), executor.getCompletedTaskCount(), executor.getQueue().size());
                } catch (Exception ex) {

                }
            }
        }, 0, 5, TimeUnit.SECONDS);

        return new Pair<>(new RpcProcessor(), executor);

    }

    private void runScanResponseTableSchedule() {
        this.timer.scheduleAtFixedRate(() -> {
            try {
                NettyServer.this.scanResponseTable();
            } catch (Exception e) {
                logger.error("scanResponseTable exception", e);
            }
        }, 1000, 3000, TimeUnit.MILLISECONDS);
    }

    private void runProcessingRequestSchedule() {
        this.processingRequestSchedule.scheduleAtFixedRate(() -> {
            try {
                scanProcessingRequest();
            } catch (Exception ex) {
                logger.error("runProcessingRequestSchedule error:{}", ex.getMessage());
            }
        }, 0, 600, TimeUnit.SECONDS);
    }

    private ServerBootstrap createServerBootstrap() {
        return this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupWorker)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.SO_LINGER, 3)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(
                                new RpcxProcessHandler(nettyServerConfig.getServerChannelMaxIdleTimeSeconds(), NettyServer.this)
                        );
                    }
                });
    }

    private String getServerPort() {
        String port = Config.ins().get("rpcx.port");
        if (StringUtils.isEmpty(port)) {
            port = "0";
        }
        return port;
    }

    private void addShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("shutdown begin");

            serverShutdown();
            shutdown();

            latch.countDown();
            logger.info("shutdown end");
        }));
    }

    private void shutdown() {
        this.eventLoopGroupWorker.shutdownGracefully();
        this.eventLoopGroupBoss.shutdownGracefully();
    }

    public void await() {
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void setGetBeanFunc(Function<Class, Object> getBeanFunc) {
        this.getBeanFunc = getBeanFunc;
    }


    public Function<Class, Object> getGetBeanFunc() {
        return getBeanFunc;
    }

    public String getAddr() {
        return addr;
    }

    public int getPort() {
        return port;
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/netty/NettyConnetManageHandler.java<|end_filename|>
package com.colobu.rpcx.netty;

import com.colobu.rpcx.common.RemotingHelper;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;

import static com.colobu.rpcx.common.RemotingUtil.closeChannel;

/**
 * @author <EMAIL>
 */
public class NettyConnetManageHandler extends ChannelDuplexHandler {

    private static final Logger log = LoggerFactory.getLogger(NettyConnetManageHandler.class);

    private final NettyRemotingAbstract nettyRemotingAbstract;

    public NettyConnetManageHandler(NettyRemotingAbstract nettyRemotingAbstract) {
        this.nettyRemotingAbstract = nettyRemotingAbstract;
    }

    @Override
    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise)
            throws Exception {
        final String local = localAddress == null ? "UNKNOW" : localAddress.toString();
        final String remote = remoteAddress == null ? "UNKNOW" : remoteAddress.toString();
        log.info("NETTY CLIENT PIPELINE: CONNECT  {} => {}", local, remote);
        super.connect(ctx, remoteAddress, localAddress, promise);

        if (nettyRemotingAbstract.hasEventListener()) {
            nettyRemotingAbstract.putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remoteAddress.toString(), ctx.channel()));
        }
    }


    @Override
    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
        log.info("NETTY CLIENT PIPELINE: DISCONNECT {}", remoteAddress);
        closeChannel(ctx.channel());
        super.disconnect(ctx, promise);

        if (nettyRemotingAbstract.hasEventListener()) {
            nettyRemotingAbstract.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress.toString(), ctx.channel()));
        }
    }


    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
        log.info("NETTY CLIENT PIPELINE: CLOSE {}", remoteAddress);
        nettyRemotingAbstract.closeChannel(ctx.channel());
        super.close(ctx, promise);

        if (nettyRemotingAbstract.hasEventListener()) {
            nettyRemotingAbstract.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress.toString(), ctx.channel()));
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent evnet = (IdleStateEvent) evt;
            if (evnet.state().equals(IdleState.ALL_IDLE)) {
                final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                log.warn("NETTY CLIENT PIPELINE: IDLE exception [{}]", remoteAddress);
                closeChannel(ctx.channel());
                if (nettyRemotingAbstract.hasEventListener()) {
                    nettyRemotingAbstract
                            .putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress.toString(), ctx.channel()));
                }
            }
        }

        ctx.fireUserEventTriggered(evt);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
        log.warn("NETTY CLIENT PIPELINE: exceptionCaught {}", remoteAddress);
        log.warn("NETTY CLIENT PIPELINE: exceptionCaught exception.", cause);
        closeChannel(ctx.channel());
        if (nettyRemotingAbstract.hasEventListener()) {
            nettyRemotingAbstract.putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress.toString(), ctx.channel()));
        }
    }

}


<|start_filename|>common/src/test/java/com/colobu/rpcx/rpc/ExporterTest.java<|end_filename|>
package com.colobu.rpcx.rpc;

import com.colobu.rpcx.rpc.impl.Exporter;
import org.junit.Test;

/**
 * Created by <EMAIL>.
 */
public class ExporterTest {


    @Test
    public void testExport() {
        System.out.println(new Exporter(null,"1:2").export("com.colobu.rpcx"));
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/TraceidFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;

import com.colobu.rpcx.common.ClassUtils;
import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcContext;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;

/**
 * @author <EMAIL>
 */
@RpcFilter(order = -999, group = {Constants.PROVIDER})
public class TraceidFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(TraceidFilter.class);

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        String clientTraceId = invocation.getAttachment(Constants.TRACE_ID);
        String key = ClassUtils.getMethodKey(invocation.getClassName(), invocation.getMethodName(), invocation.getParameterTypeNames());
        String traceId = "";
        String spanId = invocation.getAttachment(Constants.SPAN_ID, "");
        if (StringUtils.isNotEmpty(clientTraceId)) {
            traceId = clientTraceId;
            spanId = spanId + "," + UUID.randomUUID().toString();
        } else {
            //new
            traceId = UUID.randomUUID().toString();
            spanId = traceId;
        }
        logger.info("invoke begin:{} traceId:{} spanId:{}", key, traceId, spanId);
        RpcContext.getContext().getAttachments().put(Constants.TRACE_ID, traceId);
        RpcContext.getContext().getAttachments().put(Constants.SPAN_ID, spanId);
        try {
            Result res = invoker.invoke(invocation);
            logger.info("invoke end:{} success traceId:{} spanId:{}", key, traceId, spanId);
            return res;
        } catch (Throwable ex) {
            logger.info("invoke end:{} failure traceId:{} spanId:{}", key, traceId, spanId);
            throw ex;
        }
    }
}


<|start_filename|>common/src/main/java/com/colobu/rpcx/common/Bytes.java<|end_filename|>
package com.colobu.rpcx.common;

public abstract class Bytes {

    public static byte[] short2bytes(short v) {
        byte[] ret = {0, 0};
        short2bytes(v, ret);
        return ret;
    }

    public static void short2bytes(short v, byte[] b) {
        short2bytes(v, b, 0);
    }

    public static void short2bytes(short v, byte[] b, int off) {
        b[off + 1] = (byte) v;
        b[off + 0] = (byte) (v >>> 8);
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @return byte[].
     */
    public static byte[] int2bytes(int v) {
        byte[] ret = {0, 0, 0, 0};
        int2bytes(v, ret);
        return ret;
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @param b byte array.
     */
    public static void int2bytes(int v, byte[] b) {
        int2bytes(v, b, 0);
    }

    /**
     * to byte array.
     *
     * @param v   value.
     * @param b   byte array.
     * @param off array offset.
     */
    public static void int2bytes(int v, byte[] b, int off) {
        b[off + 3] = (byte) v;
        b[off + 2] = (byte) (v >>> 8);
        b[off + 1] = (byte) (v >>> 16);
        b[off + 0] = (byte) (v >>> 24);
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @return byte[].
     */
    public static byte[] float2bytes(float v) {
        byte[] ret = {0, 0, 0, 0};
        float2bytes(v, ret);
        return ret;
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @param b byte array.
     */
    public static void float2bytes(float v, byte[] b) {
        float2bytes(v, b, 0);
    }

    /**
     * to byte array.
     *
     * @param v   value.
     * @param b   byte array.
     * @param off array offset.
     */
    public static void float2bytes(float v, byte[] b, int off) {
        int i = Float.floatToIntBits(v);
        b[off + 3] = (byte) i;
        b[off + 2] = (byte) (i >>> 8);
        b[off + 1] = (byte) (i >>> 16);
        b[off + 0] = (byte) (i >>> 24);
    }


    /**
     * to byte array.
     *
     * @param v value.
     * @return byte[].
     */
    public static byte[] long2bytes(long v) {
        byte[] ret = {0, 0, 0, 0, 0, 0, 0, 0};
        long2bytes(v, ret);
        return ret;
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @param b byte array.
     */
    public static void long2bytes(long v, byte[] b) {
        long2bytes(v, b, 0);
    }

    /**
     * to byte array.
     *
     * @param v   value.
     * @param b   byte array.
     * @param off array offset.
     */
    public static void long2bytes(long v, byte[] b, int off) {
        b[off + 7] = (byte) v;
        b[off + 6] = (byte) (v >>> 8);
        b[off + 5] = (byte) (v >>> 16);
        b[off + 4] = (byte) (v >>> 24);
        b[off + 3] = (byte) (v >>> 32);
        b[off + 2] = (byte) (v >>> 40);
        b[off + 1] = (byte) (v >>> 48);
        b[off + 0] = (byte) (v >>> 56);
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @return byte[].
     */
    public static byte[] double2bytes(double v) {
        byte[] ret = {0, 0, 0, 0, 0, 0, 0, 0};
        double2bytes(v, ret);
        return ret;
    }

    /**
     * to byte array.
     *
     * @param v value.
     * @param b byte array.
     */
    public static void double2bytes(double v, byte[] b) {
        double2bytes(v, b, 0);
    }

    /**
     * to byte array.
     *
     * @param v   value.
     * @param b   byte array.
     * @param off array offset.
     */
    public static void double2bytes(double v, byte[] b, int off) {
        long j = Double.doubleToLongBits(v);
        b[off + 7] = (byte) j;
        b[off + 6] = (byte) (j >>> 8);
        b[off + 5] = (byte) (j >>> 16);
        b[off + 4] = (byte) (j >>> 24);
        b[off + 3] = (byte) (j >>> 32);
        b[off + 2] = (byte) (j >>> 40);
        b[off + 1] = (byte) (j >>> 48);
        b[off + 0] = (byte) (j >>> 56);
    }


    /**
     * to short.
     *
     * @param b byte array.
     * @return short.
     */
    public static short bytes2short(byte[] b) {
        return bytes2short(b, 0);
    }

    /**
     * to short.
     *
     * @param b   byte array.
     * @param off offset.
     * @return short.
     */
    public static short bytes2short(byte[] b, int off) {
        return (short) (((b[off + 1] & 0xFF) << 0) +
                ((b[off + 0]) << 8));
    }

    /**
     * to int.
     *
     * @param b byte array.
     * @return int.
     */
    public static int bytes2int(byte[] b) {
        return bytes2int(b, 0);
    }

    /**
     * to int.
     *
     * @param b   byte array.
     * @param off offset.
     * @return int.
     */
    public static int bytes2int(byte[] b, int off) {
        return ((b[off + 3] & 0xFF) << 0) +
                ((b[off + 2] & 0xFF) << 8) +
                ((b[off + 1] & 0xFF) << 16) +
                ((b[off + 0]) << 24);
    }

    /**
     * to int.
     *
     * @param b byte array.
     * @return int.
     */
    public static float bytes2float(byte[] b) {
        return bytes2float(b, 0);
    }

    /**
     * to int.
     *
     * @param b   byte array.
     * @param off offset.
     * @return int.
     */
    public static float bytes2float(byte[] b, int off) {
        int i = ((b[off + 3] & 0xFF) << 0) +
                ((b[off + 2] & 0xFF) << 8) +
                ((b[off + 1] & 0xFF) << 16) +
                ((b[off + 0]) << 24);
        return Float.intBitsToFloat(i);
    }

    /**
     * to long.
     *
     * @param b byte array.
     * @return long.
     */
    public static long bytes2long(byte[] b) {
        return bytes2long(b, 0);
    }

    /**
     * to long.
     *
     * @param b   byte array.
     * @param off offset.
     * @return long.
     */
    public static long bytes2long(byte[] b, int off) {
        return ((b[off + 7] & 0xFFL) << 0) +
                ((b[off + 6] & 0xFFL) << 8) +
                ((b[off + 5] & 0xFFL) << 16) +
                ((b[off + 4] & 0xFFL) << 24) +
                ((b[off + 3] & 0xFFL) << 32) +
                ((b[off + 2] & 0xFFL) << 40) +
                ((b[off + 1] & 0xFFL) << 48) +
                (((long) b[off + 0]) << 56);
    }

    /**
     * to long.
     *
     * @param b byte array.
     * @return double.
     */
    public static double bytes2double(byte[] b) {
        return bytes2double(b, 0);
    }

    /**
     * to long.
     *
     * @param b   byte array.
     * @param off offset.
     * @return double.
     */
    public static double bytes2double(byte[] b, int off) {
        long j = ((b[off + 7] & 0xFFL) << 0) +
                ((b[off + 6] & 0xFFL) << 8) +
                ((b[off + 5] & 0xFFL) << 16) +
                ((b[off + 4] & 0xFFL) << 24) +
                ((b[off + 3] & 0xFFL) << 32) +
                ((b[off + 2] & 0xFFL) << 40) +
                ((b[off + 1] & 0xFFL) << 48) +
                (((long) b[off + 0]) << 56);
        return Double.longBitsToDouble(j);
    }


}


<|start_filename|>common/src/main/java/com/colobu/rpcx/filter/impl/CacheFilter.java<|end_filename|>
package com.colobu.rpcx.filter.impl;


import com.colobu.rpcx.cache.Cache;
import com.colobu.rpcx.cache.LruCacheFactory;
import com.colobu.rpcx.common.StringUtils;
import com.colobu.rpcx.config.Constants;
import com.colobu.rpcx.filter.Filter;
import com.colobu.rpcx.rpc.Invoker;
import com.colobu.rpcx.rpc.Result;
import com.colobu.rpcx.rpc.RpcException;
import com.colobu.rpcx.rpc.annotation.RpcFilter;
import com.colobu.rpcx.rpc.impl.RpcInvocation;
import com.colobu.rpcx.rpc.impl.RpcResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author <EMAIL>
 */
@RpcFilter(order = -998, group = {Constants.PROVIDER})
public class CacheFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(CacheFilter.class);

    private static LruCacheFactory factory = new LruCacheFactory();

    @Override
    public Result invoke(Invoker<?> invoker, RpcInvocation invocation) throws RpcException {
        if (Constants.TRUE.equals(invoker.getUrl().getParameter(Constants.CACHE_KEY,Constants.FALSE))) {
            logger.debug("CacheFilter begin");
            Cache cache = factory.getCache(invoker.getUrl());
            if (cache != null) {
                String key = StringUtils.toArgumentString(invocation.getArguments());
                logger.info("cache key:{}", key);
                if (cache != null && key != null) {
                    Object value = cache.get(key);
                    if (value != null) {
                        logger.info("CacheFilter end get from cache");
                        return new RpcResult(value);
                    }
                    Result result = invoker.invoke(invocation);
                    if (!result.hasException()) {
                        cache.put(key, result.getValue());
                    }
                    logger.debug("CacheFilter end");
                    return result;
                }
            }
        }
        return invoker.invoke(invocation);
    }

}


<|start_filename|>demo/demmoserver/src/main/java/com/colobu/rpcx/service/Arith.java<|end_filename|>
package com.colobu.rpcx.service;

import com.colobu.rpcx.rpc.annotation.Provider;
import org.springframework.stereotype.Service;

/**
 * @author <EMAIL>
 */
@Provider
@Service
public class Arith implements IArith {

    /**
     * golang 调用
     */
    public byte[] Echo(byte[] params) {
        return (new String(params) + " java server!!!").getBytes();
    }

    public Integer sum(Integer a, Integer b) {
        return a + b;
    }

    public int sum2(int a, int b) {
        return a + b;
    }


}
