mirror of https://github.com/longtai-cn/hippo4j
Refactor the rpc module (#1309)
pizihao
1 year ago
committed by
GitHub
parent
53e1de5eee
commit
630e8402cd
@ -1,118 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.client;
|
||||
|
||||
import cn.hippo4j.common.toolkit.Assert;
|
||||
import cn.hippo4j.rpc.exception.TimeOutException;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import cn.hippo4j.rpc.support.NettyConnectPool;
|
||||
import cn.hippo4j.rpc.support.NettyConnectPoolHolder;
|
||||
import cn.hippo4j.rpc.support.ResultHolder;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
/**
|
||||
* Client implemented using netty
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@Slf4j
|
||||
public class NettyClientConnection implements ClientConnection {
|
||||
|
||||
InetSocketAddress address;
|
||||
/**
|
||||
* Obtain the connection timeout period. The default value is 30s
|
||||
*/
|
||||
long timeout = 30000L;
|
||||
final int nanosPerMilliSecond = 1000000;
|
||||
|
||||
EventLoopGroup worker = new NioEventLoopGroup();
|
||||
NettyConnectPool connectionPool;
|
||||
ChannelFuture future;
|
||||
Channel channel;
|
||||
|
||||
public NettyClientConnection(InetSocketAddress address,
|
||||
ChannelPoolHandler handler) {
|
||||
Assert.notNull(worker);
|
||||
this.address = address;
|
||||
this.connectionPool = NettyConnectPoolHolder.getPool(address, timeout, worker, handler);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Response connect(Request request) {
|
||||
this.channel = connectionPool.acquire(timeout);
|
||||
boolean debugEnabled = log.isDebugEnabled();
|
||||
Response response;
|
||||
try {
|
||||
String key = request.getKey();
|
||||
this.future = channel.writeAndFlush(request);
|
||||
if (debugEnabled) {
|
||||
log.debug("Call successful, target address is {}:{}, request key is {}", address.getHostName(), address.getPort(), key);
|
||||
}
|
||||
// Wait for execution to complete
|
||||
ResultHolder.putThread(key, Thread.currentThread());
|
||||
LockSupport.parkNanos(timeout() * nanosPerMilliSecond);
|
||||
response = ResultHolder.get(key);
|
||||
if (response == null) {
|
||||
throw new TimeOutException("Timeout waiting for server-side response");
|
||||
}
|
||||
if (debugEnabled) {
|
||||
log.debug("The response from {}:{} was received successfully with the response key {}.", address.getHostName(), address.getPort(), key);
|
||||
}
|
||||
return response;
|
||||
} catch (Exception ex) {
|
||||
throw ex;
|
||||
} finally {
|
||||
connectionPool.release(this.channel);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long timeout() {
|
||||
return timeout;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setTimeout(long timeout) {
|
||||
this.timeout = timeout;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
Optional.ofNullable(this.channel)
|
||||
.ifPresent(c -> {
|
||||
worker.shutdownGracefully();
|
||||
this.future.channel().close();
|
||||
this.channel.close();
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isActive() {
|
||||
return channel.isActive();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,112 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.connection;
|
||||
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import cn.hippo4j.rpc.exception.TimeOutException;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import cn.hippo4j.rpc.support.ResultHolder;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
/**
|
||||
* Client implemented using netty
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@Slf4j
|
||||
public class SimpleClientConnection implements ClientConnection {
|
||||
|
||||
InetSocketAddress address;
|
||||
/**
|
||||
* Obtain the connection timeout period. The default value is 30s
|
||||
*/
|
||||
long timeout = 30000L;
|
||||
EventLoopGroup worker = new NioEventLoopGroup();
|
||||
SimpleConnectPool connectionPool;
|
||||
static final String TIME_OUT_MSG = "Timeout waiting for server-side response";
|
||||
|
||||
public SimpleClientConnection(InetSocketAddress address,
|
||||
ChannelPoolHandler handler) {
|
||||
this.address = address;
|
||||
this.connectionPool = ConnectPoolHolder.getPool(address, timeout, worker, handler);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <R> R connect(Request request) {
|
||||
Channel channel = connectionPool.acquire(timeout);
|
||||
try {
|
||||
channel.writeAndFlush(request);
|
||||
return wait(request.getRID());
|
||||
} finally {
|
||||
connectionPool.release(channel);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* wait the Response
|
||||
*
|
||||
* @param requestId RID
|
||||
* @return Response
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <R> R wait(String requestId) {
|
||||
Response response;
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("Call successful, target address is {}:{}, request key is {}", address.getHostName(), address.getPort(), requestId);
|
||||
}
|
||||
// Wait for execution to complete
|
||||
ResultHolder.putThread(requestId, Thread.currentThread());
|
||||
LockSupport.parkNanos(timeout() * 1000000);
|
||||
response = ResultHolder.get(requestId);
|
||||
if (response == null) {
|
||||
throw new TimeOutException(TIME_OUT_MSG);
|
||||
}
|
||||
if (response.isErr()) {
|
||||
throw new ConnectionException(response.getErrMsg());
|
||||
}
|
||||
if (log.isDebugEnabled()) {
|
||||
log.debug("The response from {}:{} was received successfully with the response key {}.", address.getHostName(), address.getPort(), requestId);
|
||||
}
|
||||
return (R) response.getObj();
|
||||
}
|
||||
|
||||
@Override
|
||||
public long timeout() {
|
||||
return timeout;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setTimeout(long timeout) {
|
||||
this.timeout = timeout;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
worker.shutdownGracefully();
|
||||
connectionPool.close();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,76 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import lombok.AccessLevel;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* the registration center for Client and Server
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@NoArgsConstructor(access = AccessLevel.PRIVATE)
|
||||
public class ClassRegistry {
|
||||
|
||||
private static final Map<String, Class<?>> SERVER_REGISTER = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* get a Obj in Registry center <br>
|
||||
*
|
||||
* @param s key
|
||||
* @return t element
|
||||
*/
|
||||
public static Class<?> get(String s) {
|
||||
return SERVER_REGISTER.get(s);
|
||||
}
|
||||
|
||||
/**
|
||||
* add the element to Registry Table <br>
|
||||
* if the key already exists, failure, and return before the value of the key. <br>
|
||||
* if success return the element
|
||||
*
|
||||
* @param s key
|
||||
* @param cls element
|
||||
* @return final mapped value
|
||||
*/
|
||||
public static Class<?> set(String s, Class<?> cls) {
|
||||
return SERVER_REGISTER.putIfAbsent(s, cls);
|
||||
}
|
||||
|
||||
/**
|
||||
* add the element to Registry Table <br>
|
||||
* if the key already exists, failure, replace it
|
||||
*
|
||||
* @param s key
|
||||
* @param cls element
|
||||
*/
|
||||
public static Class<?> put(String s, Class<?> cls) {
|
||||
return SERVER_REGISTER.put(s, cls);
|
||||
}
|
||||
|
||||
/**
|
||||
* clear
|
||||
*/
|
||||
public static void clear() {
|
||||
SERVER_REGISTER.clear();
|
||||
}
|
||||
}
|
||||
@ -1,53 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ReflectUtil;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.ServiceLoader;
|
||||
|
||||
/**
|
||||
* You simply create an instance of a class based on its name and specific type.
|
||||
* Load through the ServiceLoader first. If the load fails, load directly through the instantiation.
|
||||
* If it is an interface, throw an exception. This is not elegant implementation
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
public class DefaultInstance implements Instance {
|
||||
|
||||
@Override
|
||||
public Object getInstance(Class<?> cls) {
|
||||
ServiceLoader<?> load = ServiceLoader.load(cls);
|
||||
Iterator<?> iterator = load.iterator();
|
||||
if (iterator.hasNext()) {
|
||||
return iterator.next();
|
||||
}
|
||||
return ReflectUtil.createInstance(cls);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getInstance(String name) {
|
||||
try {
|
||||
Class<?> cls = Class.forName(name);
|
||||
return getInstance(cls);
|
||||
} catch (ClassNotFoundException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,49 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.exception;
|
||||
|
||||
/**
|
||||
* HandlerNotFoundException occurs when no executable handler can be found
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
public class HandlerNotFoundException extends RuntimeException {
|
||||
|
||||
private static final long serialVersionUID = 8247610319171014183L;
|
||||
|
||||
public HandlerNotFoundException() {
|
||||
super();
|
||||
}
|
||||
|
||||
public HandlerNotFoundException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public HandlerNotFoundException(Throwable e) {
|
||||
super(e.getMessage(), e);
|
||||
}
|
||||
|
||||
public HandlerNotFoundException(String message, Throwable throwable) {
|
||||
super(message, throwable);
|
||||
}
|
||||
|
||||
public HandlerNotFoundException(String message, Throwable throwable, boolean enableSuppression, boolean writableStackTrace) {
|
||||
super(message, throwable, enableSuppression, writableStackTrace);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,61 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.exception.HandlerNotFoundException;
|
||||
import cn.hippo4j.rpc.model.DefaultResponse;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* This is a processor that does not support parameters but can get the return value. <br>
|
||||
* Even if the parameters passed by the user will not be recognized, it will even become an error
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@RequiredArgsConstructor
|
||||
@ChannelHandler.Sharable
|
||||
public class ServerBareTakeHandler<T> extends ServerHandler {
|
||||
|
||||
final String name;
|
||||
final Supplier<T> fun;
|
||||
|
||||
@Override
|
||||
String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Response sendHandler(Request request) {
|
||||
try {
|
||||
Object[] parameters = request.getParameters();
|
||||
if (parameters.length != 0) {
|
||||
throw new HandlerNotFoundException("no handler found that matches the pair " + name + " and function");
|
||||
}
|
||||
T t = fun.get();
|
||||
return new DefaultResponse(request.getRID(), t);
|
||||
} catch (Exception e) {
|
||||
return new DefaultResponse(request.getRID(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,73 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.exception.HandlerNotFoundException;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.DefaultResponse;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
||||
import java.util.function.BiFunction;
|
||||
|
||||
/**
|
||||
* netty adaptation layer about {@link DefaultRequest}<br><br>
|
||||
* Parse the parameters in the request to execute the corresponding method. <br>
|
||||
* This is a relatively flexible processor at present, but there are still great defects. <br>
|
||||
* <br>For example:<br>
|
||||
* <ul>
|
||||
* <li>This handler only supports requests with two parameters, it will not work if the number of parameters does not match</li>
|
||||
* <li>If you want to pass multiple parameters please wrap them, or customize the processor</li>
|
||||
* <li>This processor does not consider whether the types match when parsing parameters, and an error occurs if the conversion fails</li>
|
||||
* </ul>
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@ChannelHandler.Sharable
|
||||
@RequiredArgsConstructor
|
||||
public class ServerBiTakeHandler<T, P, K> extends ServerHandler {
|
||||
|
||||
final String name;
|
||||
|
||||
final BiFunction<T, P, K> fun;
|
||||
|
||||
@Override
|
||||
String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public Response sendHandler(Request request) {
|
||||
try {
|
||||
Object[] parameters = request.getParameters();
|
||||
if (parameters.length != 2) {
|
||||
throw new HandlerNotFoundException("no handler found that matches the pair " + name + " and function");
|
||||
}
|
||||
T t = (T) parameters[0];
|
||||
P p = (P) parameters[1];
|
||||
K r = fun.apply(t, p);
|
||||
return new DefaultResponse(request.getRID(), r);
|
||||
} catch (Exception e) {
|
||||
return new DefaultResponse(request.getRID(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,55 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* The handler located on the server side provides unified operations for the server side
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
abstract class ServerHandler extends AbstractTakeHandler {
|
||||
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
||||
if (!(msg instanceof DefaultRequest)) {
|
||||
ctx.fireChannelRead(msg);
|
||||
return;
|
||||
}
|
||||
Request request = (Request) msg;
|
||||
if (!Objects.equals(request.getKey(), getName())) {
|
||||
ctx.fireChannelRead(msg);
|
||||
return;
|
||||
}
|
||||
Response response = sendHandler(request);
|
||||
ctx.writeAndFlush(response);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the name of the current handler
|
||||
*
|
||||
* @return name
|
||||
*/
|
||||
abstract String getName();
|
||||
}
|
||||
@ -0,0 +1,71 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.exception.HandlerNotFoundException;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.DefaultResponse;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* netty adaptation layer about {@link DefaultRequest}<br><br>
|
||||
* Parse the parameters in the request to execute the corresponding method. <br>
|
||||
* This is a relatively flexible processor at present, but there are still great defects. <br>
|
||||
* <br>For example:<br>
|
||||
* <ul>
|
||||
* <li>This handler only supports requests with one parameters, it will not work if the number of parameters does not match</li>
|
||||
* <li>If you want to pass multiple parameters please wrap them, or customize the processor</li>
|
||||
* <li>This processor does not consider whether the types match when parsing parameters, and an error occurs if the conversion fails</li>
|
||||
* </ul>
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@ChannelHandler.Sharable
|
||||
@RequiredArgsConstructor
|
||||
public class ServerTakeHandler<T, R> extends ServerHandler {
|
||||
|
||||
final String name;
|
||||
final Function<T, R> fun;
|
||||
|
||||
@Override
|
||||
String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public Response sendHandler(Request request) {
|
||||
try {
|
||||
Object[] parameters = request.getParameters();
|
||||
if (parameters.length != 1) {
|
||||
throw new HandlerNotFoundException("no handler found that matches the pair " + name + " and function");
|
||||
}
|
||||
T t = (T) parameters[0];
|
||||
R r = fun.apply(t);
|
||||
return new DefaultResponse(request.getRID(), r);
|
||||
} catch (Exception e) {
|
||||
return new DefaultResponse(request.getRID(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import lombok.AccessLevel;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
@NoArgsConstructor(access = AccessLevel.PRIVATE)
|
||||
public class AddressUtil {
|
||||
|
||||
private static final String HTTP = "http://";
|
||||
private static final String HTTPS = "https://";
|
||||
|
||||
/**
|
||||
* parsing hostname
|
||||
*
|
||||
* @param address address
|
||||
* @return InetAddress
|
||||
*/
|
||||
public static InetSocketAddress getInetAddress(String address) {
|
||||
if (address.startsWith(HTTP)) {
|
||||
address = address.replaceFirst(HTTP, "");
|
||||
}
|
||||
|
||||
if (address.startsWith(HTTPS)) {
|
||||
address = address.replaceFirst(HTTPS, "");
|
||||
}
|
||||
|
||||
String[] addressStr = address.split(":");
|
||||
if (addressStr.length < 2) {
|
||||
throw new ConnectionException("Failed to connect to the server because the IP address is invalid. Procedure");
|
||||
}
|
||||
return InetSocketAddress.createUnresolved(addressStr[0], Integer.parseInt(addressStr[1]));
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,140 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.common.toolkit.IdUtil;
|
||||
import cn.hippo4j.rpc.client.Client;
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import cn.hippo4j.rpc.handler.NettyClientPoolHandler;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import lombok.AccessLevel;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
import java.lang.reflect.InvocationHandler;
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* Add a proxy for the request, {@link Proxy} and {@link InvocationHandler}
|
||||
*
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@NoArgsConstructor(access = AccessLevel.PRIVATE)
|
||||
public class NettyProxyCenter {
|
||||
|
||||
// cache
|
||||
static Map<String, Object> map = new ConcurrentHashMap<>();
|
||||
|
||||
/**
|
||||
* A proxy object for PRC is obtained through an interface
|
||||
*
|
||||
* @param cls The interface type
|
||||
* @param address address
|
||||
* @param <T> Object type
|
||||
* @param handler the pool handler for netty
|
||||
* @return Proxy objects
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T getProxy(Class<T> cls, InetSocketAddress address, NettyClientPoolHandler handler) {
|
||||
Client client = NettyClientSupport.getClient(address, handler);
|
||||
String s = address + cls.getName();
|
||||
Object o = map.get(s);
|
||||
if (o != null) {
|
||||
return (T) o;
|
||||
}
|
||||
return createProxy(client, cls, address);
|
||||
}
|
||||
|
||||
/**
|
||||
* A proxy object for PRC is obtained through an interface
|
||||
*
|
||||
* @param cls The interface type
|
||||
* @param address address String
|
||||
* @param <T> Object type
|
||||
* @return Proxy objects
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T getProxy(Class<T> cls, String address) {
|
||||
String[] addressStr = address.split(":");
|
||||
if (addressStr.length < 2) {
|
||||
throw new ConnectionException("Failed to connect to the server because the IP address is invalid. Procedure");
|
||||
}
|
||||
InetSocketAddress socketAddress = InetSocketAddress.createUnresolved(addressStr[0], Integer.parseInt(addressStr[1]));
|
||||
String s = socketAddress + cls.getName();
|
||||
Object o = map.get(s);
|
||||
if (o != null) {
|
||||
return (T) o;
|
||||
}
|
||||
Client client = NettyClientSupport.getClient(socketAddress);
|
||||
return createProxy(client, cls, socketAddress);
|
||||
}
|
||||
|
||||
/**
|
||||
* remove proxy object
|
||||
*
|
||||
* @param cls the class
|
||||
* @param address address String
|
||||
*/
|
||||
public static void removeProxy(Class<?> cls, String address) {
|
||||
String[] addressStr = address.split(":");
|
||||
if (addressStr.length < 2) {
|
||||
throw new ConnectionException("Failed to connect to the server because the IP address is invalid. Procedure");
|
||||
}
|
||||
InetSocketAddress socketAddress = InetSocketAddress.createUnresolved(addressStr[0], Integer.parseInt(addressStr[1]));
|
||||
String s = socketAddress + cls.getName();
|
||||
NettyClientSupport.closeClient(socketAddress);
|
||||
map.remove(s);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T> T createProxy(Client client, Class<T> cls, InetSocketAddress address) {
|
||||
boolean b = cls.isInterface();
|
||||
if (!b) {
|
||||
throw new RuntimeException(cls.getName() + "is not a Interface");
|
||||
}
|
||||
String s = address.toString() + cls.getName();
|
||||
Object o = map.get(s);
|
||||
if (o != null) {
|
||||
return (T) o;
|
||||
}
|
||||
T obj = (T) Proxy.newProxyInstance(
|
||||
cls.getClassLoader(),
|
||||
new Class[]{cls},
|
||||
(proxy, method, args) -> {
|
||||
String clsName = cls.getName();
|
||||
String methodName = method.getName();
|
||||
String key = address + clsName + methodName + IdUtil.simpleUUID();
|
||||
Class<?>[] parameterTypes = method.getParameterTypes();
|
||||
Request request = new DefaultRequest(key, clsName, methodName, parameterTypes, args);
|
||||
Response response = client.connection(request);
|
||||
if (response == null) {
|
||||
return null;
|
||||
}
|
||||
if (response.isErr()) {
|
||||
throw new RuntimeException(response.getErrMsg(), response.getThrowable());
|
||||
}
|
||||
return response.getObj();
|
||||
});
|
||||
map.put(s, obj);
|
||||
return obj;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,95 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.client;
|
||||
|
||||
import cn.hippo4j.rpc.connection.ClientConnection;
|
||||
import cn.hippo4j.rpc.connection.ServerConnection;
|
||||
import cn.hippo4j.rpc.connection.SimpleClientConnection;
|
||||
import cn.hippo4j.rpc.connection.SimpleServerConnection;
|
||||
import cn.hippo4j.rpc.handler.ServerBareTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.ServerBiTakeHandler;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.handler.ClientPoolHandler;
|
||||
import cn.hippo4j.rpc.handler.ClientTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.ServerTakeHandler;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
public class ClientSupportTest {
|
||||
|
||||
static ServerPort port = new TestServerPort();
|
||||
static final String addressStr = "localhost";
|
||||
|
||||
static final String take = "serverTake";
|
||||
static final String biTake = "biTake";
|
||||
static final String bareTake = "bareTake";
|
||||
static final String timeout = "timeout";
|
||||
static RPCServer rpcServer;
|
||||
|
||||
@BeforeClass
|
||||
public static void startServer() {
|
||||
CallManager manager = new CallManager();
|
||||
ServerTakeHandler<Integer, Integer> takeHandler = new ServerTakeHandler<>(biTake, manager::call);
|
||||
ServerBiTakeHandler<Integer, Integer, Integer> biTakeHandler = new ServerBiTakeHandler<>(take, manager::call);
|
||||
ServerBareTakeHandler<Integer> bareTakeHandler = new ServerBareTakeHandler<>(bareTake, manager::call);
|
||||
ServerBareTakeHandler<Integer> timeoutHandler = new ServerBareTakeHandler<>(timeout, manager::callTestTimeout);
|
||||
ServerConnection connection = new SimpleServerConnection(takeHandler, bareTakeHandler, biTakeHandler, timeoutHandler);
|
||||
rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
|
||||
}
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void stopServer() throws IOException {
|
||||
if (rpcServer.isActive()) {
|
||||
rpcServer.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void closeTest() throws IOException {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(addressStr, port.getPort());
|
||||
ChannelPoolHandler channelPoolHandler = new ClientPoolHandler(new ClientTakeHandler());
|
||||
ClientConnection clientConnection = new SimpleClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
|
||||
ClientSupport.closeClient(new InetSocketAddress(addressStr, port.getPort()));
|
||||
rpcClient.close();
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,161 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.client;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.discovery.ClassRegistry;
|
||||
import cn.hippo4j.rpc.discovery.DefaultInstance;
|
||||
import cn.hippo4j.rpc.discovery.Instance;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.handler.NettyClientPoolHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyClientTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyServerTakeHandler;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import cn.hippo4j.rpc.server.NettyServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import cn.hippo4j.rpc.server.ServerConnection;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
public class RPCClientTest {
|
||||
|
||||
String host = "localhost";
|
||||
ServerPort port = new TestServerPort();
|
||||
ServerPort portTest = new TestPortServerPort();
|
||||
|
||||
/**
|
||||
* This test case can be overridden under the handler and coder packages
|
||||
*/
|
||||
@Test
|
||||
public void connection() throws IOException {
|
||||
Class<CallManager> cls = CallManager.class;
|
||||
String className = cls.getName();
|
||||
ClassRegistry.put(className, cls);
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, portTest);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, portTest.getPort());
|
||||
ChannelPoolHandler channelPoolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
NettyClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
Class<?>[] classes = new Class<?>[2];
|
||||
classes[0] = Integer.class;
|
||||
classes[1] = Integer.class;
|
||||
Object[] objects = new Object[2];
|
||||
objects[0] = 1;
|
||||
objects[1] = 2;
|
||||
Request request = new DefaultRequest("127.0.0.18889", className, "callTest", classes, objects);
|
||||
Response response = rpcClient.connection(request);
|
||||
boolean active = rpcClient.isActive();
|
||||
Assert.assertTrue(active);
|
||||
Assert.assertEquals(response.getObj(), 3);
|
||||
rpcClient.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectionTest() throws IOException {
|
||||
Class<CallManager> cls = CallManager.class;
|
||||
String className = cls.getName();
|
||||
ClassRegistry.put(className, cls);
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
ChannelPoolHandler channelPoolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
ClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
Request request = new DefaultRequest("127.0.0.18888", className, "call", null, null);
|
||||
for (int i = 0; i < 50; i++) {
|
||||
Response response = rpcClient.connection(request);
|
||||
boolean active = rpcClient.isActive();
|
||||
Assert.assertTrue(active);
|
||||
Assert.assertEquals(response.getObj(), 1);
|
||||
}
|
||||
rpcClient.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
@Test(expected = Exception.class)
|
||||
public void responseNullExceptionTest() throws IOException {
|
||||
Class<CallManager> cls = CallManager.class;
|
||||
String className = cls.getName();
|
||||
ClassRegistry.put(className, cls);
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
ChannelPoolHandler channelPoolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
ClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
clientConnection.setTimeout(300L);
|
||||
try (RPCClient rpcClient = new RPCClient(clientConnection)) {
|
||||
Request request = new DefaultRequest("127.0.0.18888", className, "callTestTimeout", null, null);
|
||||
Response response = rpcClient.connection(request);
|
||||
Assert.assertNotNull(response.getErrMsg());
|
||||
Assert.assertNotNull(response.getThrowable());
|
||||
} catch (IOException e) {
|
||||
// no something
|
||||
} finally {
|
||||
rpcServer.close();
|
||||
}
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
|
||||
static class TestPortServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,67 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
public class ClassRegistryTest {
|
||||
|
||||
@Test
|
||||
public void get() {
|
||||
String getStr = "GetModel";
|
||||
Class<?> cls = ClassRegistry.get(getStr);
|
||||
Assert.assertNull(cls);
|
||||
ClassRegistry.put(getStr, GetModel.class);
|
||||
Class<?> aClass = ClassRegistry.get(getStr);
|
||||
Assert.assertNotNull(aClass);
|
||||
ClassRegistry.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void set() {
|
||||
String getStr = "GetModel";
|
||||
ClassRegistry.set(getStr, GetModel.class);
|
||||
Class<?> aClass = ClassRegistry.get(getStr);
|
||||
Assert.assertEquals(aClass, GetModel.class);
|
||||
ClassRegistry.set(getStr, SetModel.class);
|
||||
Class<?> aClass1 = ClassRegistry.get(getStr);
|
||||
Assert.assertEquals(aClass1, GetModel.class);
|
||||
ClassRegistry.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void put() {
|
||||
String getStr = "GetModel";
|
||||
ClassRegistry.put(getStr, GetModel.class);
|
||||
Class<?> aClass = ClassRegistry.get(getStr);
|
||||
Assert.assertEquals(aClass, GetModel.class);
|
||||
ClassRegistry.put(getStr, SetModel.class);
|
||||
Class<?> aClass1 = ClassRegistry.get(getStr);
|
||||
Assert.assertEquals(aClass1, SetModel.class);
|
||||
ClassRegistry.clear();
|
||||
}
|
||||
|
||||
public static class GetModel {
|
||||
|
||||
}
|
||||
|
||||
public static class SetModel {
|
||||
|
||||
}
|
||||
}
|
||||
@ -1,59 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
public class DefaultInstanceTest {
|
||||
|
||||
Instance instance = new DefaultInstance();
|
||||
|
||||
@Test
|
||||
public void getInstance() {
|
||||
Class<InstanceModel> cls = InstanceModel.class;
|
||||
Object instanceInstance = instance.getInstance(cls);
|
||||
Assert.assertNotNull(instanceInstance);
|
||||
Assert.assertEquals(cls, instanceInstance.getClass());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGetInstance() {
|
||||
String className = "cn.hippo4j.rpc.discovery.InstanceModel";
|
||||
Object instanceInstance = instance.getInstance(className);
|
||||
Assert.assertNotNull(instanceInstance);
|
||||
Assert.assertEquals(className, instanceInstance.getClass().getName());
|
||||
}
|
||||
|
||||
@Test(expected = RuntimeException.class)
|
||||
public void testGetInstanceTest() {
|
||||
String className = "cn.hippo4j.rpc.discovery.InstanceModelTest";
|
||||
Object instanceInstance = instance.getInstance(className);
|
||||
Assert.assertNotNull(instanceInstance);
|
||||
Assert.assertEquals(className, instanceInstance.getClass().getName());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void getInstanceTest() {
|
||||
Class<InstanceServerLoader> cls = InstanceServerLoader.class;
|
||||
Object instanceInstance = instance.getInstance(cls);
|
||||
Assert.assertNotNull(instanceInstance);
|
||||
Assert.assertEquals(InstanceServerLoaderImpl.class, instanceInstance.getClass());
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,34 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.Setter;
|
||||
import org.springframework.boot.test.context.TestComponent;
|
||||
|
||||
@Setter
|
||||
@Getter
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
@TestComponent
|
||||
public class InstanceModel {
|
||||
|
||||
String name;
|
||||
}
|
||||
@ -1,24 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
public interface InstanceServerLoader {
|
||||
|
||||
String getName();
|
||||
|
||||
}
|
||||
@ -1,33 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.Setter;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class InstanceServerLoaderImpl implements InstanceServerLoader {
|
||||
|
||||
String name = "name";
|
||||
|
||||
}
|
||||
@ -1,48 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.discovery;
|
||||
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
/**
|
||||
* TODO Common module removes spring dependency leftovers
|
||||
*/
|
||||
// @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = {InstanceModel.class})
|
||||
// @RunWith(SpringJUnit4ClassRunner.class)
|
||||
public class SpringContextInstanceTest {
|
||||
|
||||
Instance instance = new SpringContextInstance();
|
||||
|
||||
// @Test
|
||||
public void getInstance() {
|
||||
Object obj = instance.getInstance(InstanceModel.class);
|
||||
Assert.assertNotNull(obj);
|
||||
Assert.assertEquals(obj.getClass(), InstanceModel.class);
|
||||
}
|
||||
|
||||
// @Test
|
||||
public void testGetInstance() {
|
||||
Object obj = instance.getInstance("instanceModel");
|
||||
Assert.assertNotNull(obj);
|
||||
Assert.assertEquals(obj.getClass(), InstanceModel.class);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,95 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.exception.OperationException;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
public class ClientPoolHandlerTest {
|
||||
|
||||
static final String test = "Test";
|
||||
static final String test1 = "Test1";
|
||||
|
||||
@Test
|
||||
public void testGetHandlerEntity() {
|
||||
TestHandler handler = new TestHandler();
|
||||
long order = 0;
|
||||
String name = test;
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler();
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity = poolHandler.getHandlerEntity(order, handler, name);
|
||||
Assert.assertEquals(entity.getName(), name);
|
||||
Assert.assertEquals(entity.getOrder(), order);
|
||||
Assert.assertEquals(entity.getHandler(), handler);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCompareTo() {
|
||||
TestHandler handler = new TestHandler();
|
||||
long order = 0;
|
||||
TestHandler handler1 = new TestHandler();
|
||||
long order1 = 1;
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler();
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity = poolHandler.getHandlerEntity(order, handler, test);
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity1 = poolHandler.getHandlerEntity(order1, handler1, test1);
|
||||
int compare = entity.compareTo(entity1);
|
||||
Assert.assertTrue(compare < 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addLast() {
|
||||
ClientPoolHandler handler = new ClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addLast(null, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addFirst() {
|
||||
ClientPoolHandler handler = new ClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addFirst(null, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddLast() {
|
||||
ClientPoolHandler handler = new ClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addLast(test, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddFirst() {
|
||||
ClientPoolHandler handler = new ClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addFirst(test, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test(expected = OperationException.class)
|
||||
public void testGetHandlerEntityFalse() {
|
||||
TestFalseHandler handler = new TestFalseHandler();
|
||||
long order = 0;
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler();
|
||||
poolHandler.getHandlerEntity(order, handler, test);
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,155 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.client.CallManager;
|
||||
import cn.hippo4j.rpc.client.ClientConnection;
|
||||
import cn.hippo4j.rpc.client.NettyClientConnection;
|
||||
import cn.hippo4j.rpc.client.RPCClient;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.discovery.ClassRegistry;
|
||||
import cn.hippo4j.rpc.discovery.DefaultInstance;
|
||||
import cn.hippo4j.rpc.discovery.Instance;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.model.Response;
|
||||
import cn.hippo4j.rpc.server.NettyServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import cn.hippo4j.rpc.server.ServerConnection;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class NettyClientPoolHandlerTest {
|
||||
|
||||
@Test
|
||||
public void testGetHandlerEntity() {
|
||||
TestHandler handler = new TestHandler();
|
||||
long order = 0;
|
||||
String name = "Test";
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler();
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity = poolHandler.getHandlerEntity(order, handler, name);
|
||||
Assert.assertEquals(entity.getName(), name);
|
||||
Assert.assertEquals(entity.getOrder(), order);
|
||||
Assert.assertEquals(entity.getHandler(), handler);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCompareTo() {
|
||||
TestHandler handler = new TestHandler();
|
||||
long order = 0;
|
||||
String name = "Test";
|
||||
TestHandler handler1 = new TestHandler();
|
||||
long order1 = 1;
|
||||
String name1 = "Test1";
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler();
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity = poolHandler.getHandlerEntity(order, handler, name);
|
||||
HandlerManager.HandlerEntity<ChannelHandler> entity1 = poolHandler.getHandlerEntity(order1, handler1, name1);
|
||||
int compare = entity.compareTo(entity1);
|
||||
Assert.assertTrue(compare < 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addLast() {
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addLast(null, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void addFirst() {
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addFirst(null, new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddLast() {
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addLast("Test", new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddFirst() {
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler();
|
||||
Assert.assertTrue(handler.isEmpty());
|
||||
handler.addFirst("Test", new TestHandler());
|
||||
Assert.assertFalse(handler.isEmpty());
|
||||
}
|
||||
|
||||
@Test(expected = RuntimeException.class)
|
||||
public void testGetHandlerEntityFalse() {
|
||||
TestFalseHandler handler = new TestFalseHandler();
|
||||
long order = 0;
|
||||
String name = "Test";
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler();
|
||||
poolHandler.getHandlerEntity(order, handler, name);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectionTest() throws IOException {
|
||||
ServerPort port = new ServerPort() {
|
||||
|
||||
final int a = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return a;
|
||||
}
|
||||
};
|
||||
Class<CallManager> cls = CallManager.class;
|
||||
String className = cls.getName();
|
||||
ClassRegistry.put(className, cls);
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", port.getPort());
|
||||
List<ChannelHandler> handlers = new ArrayList<>();
|
||||
handlers.add(new NettyClientTakeHandler());
|
||||
NettyClientPoolHandler channelPoolHandler = new NettyClientPoolHandler(handlers);
|
||||
channelPoolHandler.addLast("test", new TestHandler());
|
||||
ClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
Request request = new DefaultRequest("127.0.0.18888", className, "call", null, null);
|
||||
for (int i = 0; i < 50; i++) {
|
||||
Response response = rpcClient.connection(request);
|
||||
boolean active = rpcClient.isActive();
|
||||
Assert.assertTrue(active);
|
||||
Assert.assertEquals(response.getObj(), 1);
|
||||
}
|
||||
rpcClient.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,141 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.client.CallManager;
|
||||
import cn.hippo4j.rpc.client.ClientSupport;
|
||||
import cn.hippo4j.rpc.client.RPCClient;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.connection.ServerConnection;
|
||||
import cn.hippo4j.rpc.connection.SimpleClientConnection;
|
||||
import cn.hippo4j.rpc.connection.SimpleServerConnection;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import cn.hippo4j.rpc.model.DefaultRequest;
|
||||
import cn.hippo4j.rpc.model.Request;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import cn.hippo4j.rpc.support.AddressUtil;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Assert;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
public class ServerHandlerTest {
|
||||
|
||||
static final String host = "localhost";
|
||||
static ServerPort port = new TestServerPort();
|
||||
|
||||
static final String take = "serverTakeServer";
|
||||
static final String biTake = "biTakeServer";
|
||||
static final String bareTake = "bareTakeServer";
|
||||
static final String timeout = "timeoutServer";
|
||||
static final String error = "errorServer";
|
||||
static RPCServer rpcServer;
|
||||
|
||||
@BeforeClass
|
||||
public static void startServer() {
|
||||
CallManager manager = new CallManager();
|
||||
ServerTakeHandler<Integer, Integer> takeHandler = new ServerTakeHandler<>(take, manager::call);
|
||||
ServerBiTakeHandler<Integer, Integer, Integer> biTakeHandler = new ServerBiTakeHandler<>(biTake, manager::call);
|
||||
ServerBareTakeHandler<Integer> bareTakeHandler = new ServerBareTakeHandler<>(bareTake, manager::call);
|
||||
ServerBareTakeHandler<Integer> timeoutHandler = new ServerBareTakeHandler<>(timeout, manager::callTestTimeout);
|
||||
ErrorServerHandler error = new ErrorServerHandler();
|
||||
ServerConnection connection = new SimpleServerConnection(takeHandler, bareTakeHandler, biTakeHandler, timeoutHandler, error);
|
||||
rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
|
||||
}
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void stopServer() throws IOException {
|
||||
if (rpcServer.isActive()) {
|
||||
rpcServer.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* This test case can be overridden under the handler and coder packages
|
||||
*/
|
||||
@Test
|
||||
public void connection() {
|
||||
String s = host + ":" + port.getPort();
|
||||
int send = ClientSupport.clientSend(s, take, 1);
|
||||
Assert.assertEquals(send, 1);
|
||||
InetSocketAddress socketAddress = AddressUtil.getInetAddress(s);
|
||||
ClientSupport.closeClient(socketAddress);
|
||||
}
|
||||
|
||||
@Test(expected = ConnectionException.class)
|
||||
public void connectionError() {
|
||||
String s = host + ":" + port.getPort();
|
||||
int send = ClientSupport.clientSend(s, error, 1);
|
||||
Assert.assertEquals(1, send);
|
||||
InetSocketAddress socketAddress = AddressUtil.getInetAddress(s);
|
||||
ClientSupport.closeClient(socketAddress);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectionTest() {
|
||||
Integer[] params = {1, 6};
|
||||
String s = host + ":" + port.getPort();
|
||||
int send = ClientSupport.clientSend(s, biTake, params);
|
||||
Assert.assertEquals(7, send);
|
||||
InetSocketAddress socketAddress = AddressUtil.getInetAddress(s);
|
||||
ClientSupport.closeClient(socketAddress);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void connectionTestBare() {
|
||||
String s = host + ":" + port.getPort();
|
||||
int send = ClientSupport.clientSend(s, bareTake);
|
||||
Assert.assertEquals(1, send);
|
||||
InetSocketAddress socketAddress = AddressUtil.getInetAddress(s);
|
||||
ClientSupport.closeClient(socketAddress);
|
||||
}
|
||||
|
||||
@Test(expected = Exception.class)
|
||||
public void responseNullExceptionTest() {
|
||||
String s = host + ":" + port.getPort();
|
||||
ClientPoolHandler handler = new ClientPoolHandler(new ClientTakeHandler());
|
||||
InetSocketAddress socketAddress = AddressUtil.getInetAddress(s);
|
||||
SimpleClientConnection connection = new SimpleClientConnection(socketAddress, handler);
|
||||
connection.setTimeout(1L);
|
||||
RPCClient client = new RPCClient(connection);
|
||||
Request request = new DefaultRequest(UUID.randomUUID().toString(), timeout);
|
||||
client.connect(request);
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,119 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.handler;
|
||||
|
||||
import cn.hippo4j.rpc.client.CallManager;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.connection.SimpleConnectPool;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.connection.SimpleServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import cn.hippo4j.rpc.connection.ServerConnection;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.util.concurrent.Future;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Assert;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.locks.LockSupport;
|
||||
|
||||
public class SimpleConnectPoolTest {
|
||||
|
||||
String host = "127.0.0.1";
|
||||
int maxCount = 64;
|
||||
int timeout = 5000;
|
||||
EventLoopGroup group = new NioEventLoopGroup();
|
||||
Class<? extends Channel> cls = NioSocketChannel.class;
|
||||
|
||||
static ServerPort port = new TestServerPort();
|
||||
|
||||
static final String take = "serverTake";
|
||||
static final String biTake = "biTake";
|
||||
static final String bareTake = "bareTake";
|
||||
static final String timeoutTake = "timeout";
|
||||
static RPCServer rpcServer;
|
||||
|
||||
@BeforeClass
|
||||
public static void startServer() {
|
||||
CallManager manager = new CallManager();
|
||||
ServerTakeHandler<Integer, Integer> takeHandler = new ServerTakeHandler<>(biTake, manager::call);
|
||||
ServerBiTakeHandler<Integer, Integer, Integer> biTakeHandler = new ServerBiTakeHandler<>(take, manager::call);
|
||||
ServerBareTakeHandler<Integer> bareTakeHandler = new ServerBareTakeHandler<>(bareTake, manager::call);
|
||||
ServerBareTakeHandler<Integer> timeoutHandler = new ServerBareTakeHandler<>(timeoutTake, manager::callTestTimeout);
|
||||
ServerConnection connection = new SimpleServerConnection(takeHandler, bareTakeHandler, biTakeHandler, timeoutHandler);
|
||||
rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
|
||||
}
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void stopServer() throws IOException {
|
||||
if (rpcServer.isActive()) {
|
||||
rpcServer.close();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void acquire() {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler(new ClientTakeHandler());
|
||||
SimpleConnectPool pool = new SimpleConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Channel acquire = pool.acquire(timeout);
|
||||
Assert.assertNotNull(acquire);
|
||||
pool.release(acquire);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAcquire() {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler(new ClientTakeHandler());
|
||||
SimpleConnectPool pool = new SimpleConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Future<Channel> acquire = pool.acquire();
|
||||
Assert.assertNotNull(acquire);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void close() {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
ClientPoolHandler poolHandler = new ClientPoolHandler(new ClientTakeHandler());
|
||||
SimpleConnectPool pool = new SimpleConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Channel acquire = pool.acquire(timeout);
|
||||
Assert.assertNotNull(acquire);
|
||||
pool.release(acquire);
|
||||
pool.close();
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,52 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
public class AddressUtilTest {
|
||||
|
||||
String address1 = "http://hippo4j.cn/login:8080";
|
||||
String address2 = "https://hippo4j.cn/login:8080";
|
||||
String address3 = "https://hippo4j.cn/login";
|
||||
String addressHostName = "hippo4j.cn/login";
|
||||
int addressPort = 8080;
|
||||
|
||||
@Test
|
||||
public void test() {
|
||||
InetSocketAddress address = AddressUtil.getInetAddress(address1);
|
||||
Assert.assertEquals(addressHostName, address.getHostName());
|
||||
Assert.assertEquals(addressPort, address.getPort());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAddress2() {
|
||||
InetSocketAddress address = AddressUtil.getInetAddress(address2);
|
||||
Assert.assertEquals(addressHostName, address.getHostName());
|
||||
Assert.assertEquals(addressPort, address.getPort());
|
||||
}
|
||||
|
||||
@Test(expected = ConnectionException.class)
|
||||
public void testAddress3() {
|
||||
AddressUtil.getInetAddress(address3);
|
||||
}
|
||||
}
|
||||
@ -1,72 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.client.CallManager;
|
||||
import cn.hippo4j.rpc.client.ClientConnection;
|
||||
import cn.hippo4j.rpc.client.NettyClientConnection;
|
||||
import cn.hippo4j.rpc.client.RPCClient;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.discovery.ClassRegistry;
|
||||
import cn.hippo4j.rpc.discovery.DefaultInstance;
|
||||
import cn.hippo4j.rpc.discovery.Instance;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.handler.NettyClientPoolHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyClientTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyServerTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.TestHandler;
|
||||
import cn.hippo4j.rpc.server.NettyServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
public class NettyClientSupportTest {
|
||||
|
||||
@Test
|
||||
public void closeTest() throws IOException {
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
ServerPort serverPort = () -> port;
|
||||
Class<CallManager> cls = CallManager.class;
|
||||
String className = cls.getName();
|
||||
ClassRegistry.put(className, cls);
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
NettyServerConnection connection = new NettyServerConnection(handler);
|
||||
connection.addLast("Test", new TestHandler());
|
||||
RPCServer rpcServer = new RPCServer(connection, serverPort);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", port);
|
||||
ChannelPoolHandler channelPoolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
ClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
|
||||
NettyClientSupport.closeClient(new InetSocketAddress("localhost", port));
|
||||
|
||||
rpcClient.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,122 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.discovery.DefaultInstance;
|
||||
import cn.hippo4j.rpc.discovery.Instance;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.handler.NettyClientPoolHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyClientTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyServerTakeHandler;
|
||||
import cn.hippo4j.rpc.server.NettyServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import cn.hippo4j.rpc.server.ServerConnection;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.nio.NioSocketChannel;
|
||||
import io.netty.util.concurrent.Future;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
public class NettyConnectPoolTest {
|
||||
|
||||
String host = "127.0.0.1";
|
||||
ServerPort port = new TestServerPort();
|
||||
int maxCount = 64;
|
||||
int timeout = 5000;
|
||||
EventLoopGroup group = new NioEventLoopGroup();
|
||||
Class<? extends Channel> cls = NioSocketChannel.class;
|
||||
|
||||
@Test
|
||||
public void acquire() throws IOException {
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
// Given the delay in starting the server, wait here
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
NettyConnectPool pool = new NettyConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Channel acquire = pool.acquire(timeout);
|
||||
Assert.assertNotNull(acquire);
|
||||
pool.release(acquire);
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAcquire() throws IOException {
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
// Given the delay in starting the server, wait here
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
NettyConnectPool pool = new NettyConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Future<Channel> acquire = pool.acquire();
|
||||
Assert.assertNotNull(acquire);
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void close() throws IOException {
|
||||
// The mode connection was denied when the server was started on the specified port
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler handler = new NettyServerTakeHandler(instance);
|
||||
ServerConnection connection = new NettyServerConnection(handler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
// Given the delay in starting the server, wait here
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved(host, port.getPort());
|
||||
NettyClientPoolHandler poolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
NettyConnectPool pool = new NettyConnectPool(address, maxCount, timeout, group, cls, poolHandler);
|
||||
Channel acquire = pool.acquire(timeout);
|
||||
Assert.assertNotNull(acquire);
|
||||
pool.release(acquire);
|
||||
pool.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,134 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.client.Client;
|
||||
import cn.hippo4j.rpc.client.NettyClientConnection;
|
||||
import cn.hippo4j.rpc.client.RPCClient;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.discovery.ClassRegistry;
|
||||
import cn.hippo4j.rpc.discovery.DefaultInstance;
|
||||
import cn.hippo4j.rpc.discovery.Instance;
|
||||
import cn.hippo4j.rpc.discovery.InstanceServerLoader;
|
||||
import cn.hippo4j.rpc.discovery.ServerPort;
|
||||
import cn.hippo4j.rpc.exception.ConnectionException;
|
||||
import cn.hippo4j.rpc.handler.NettyClientPoolHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyClientTakeHandler;
|
||||
import cn.hippo4j.rpc.handler.NettyServerTakeHandler;
|
||||
import cn.hippo4j.rpc.server.NettyServerConnection;
|
||||
import cn.hippo4j.rpc.server.RPCServer;
|
||||
import io.netty.channel.pool.ChannelPoolHandler;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
public class NettyProxyCenterTest {
|
||||
|
||||
ServerPort port = new TestServerPort();
|
||||
|
||||
@Test
|
||||
public void getProxy() {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", port.getPort());
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
ProxyInterface localhost = NettyProxyCenter.getProxy(ProxyInterface.class, address, handler);
|
||||
Assert.assertNotNull(localhost);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void createProxy() {
|
||||
ProxyInterface localhost = NettyProxyCenter.getProxy(ProxyInterface.class, "localhost:8894");
|
||||
Assert.assertNotNull(localhost);
|
||||
NettyProxyCenter.getProxy(ProxyInterface.class, "localhost:8894");
|
||||
InetSocketAddress socketAddress = InetSocketAddress.createUnresolved("localhost", 8894);
|
||||
Client client = NettyClientSupport.getClient(socketAddress);
|
||||
ProxyInterface proxy = NettyProxyCenter.createProxy(client, ProxyInterface.class, socketAddress);
|
||||
Assert.assertNotNull(proxy);
|
||||
}
|
||||
|
||||
@Test(expected = ConnectionException.class)
|
||||
public void createProxyException() {
|
||||
NettyProxyCenter.getProxy(ProxyInterface.class, "localhost8894");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void removeProxy() {
|
||||
NettyProxyCenter.getProxy(ProxyInterface.class, "localhost:8894");
|
||||
NettyProxyCenter.removeProxy(ProxyInterface.class, "localhost:8894");
|
||||
}
|
||||
|
||||
@Test(expected = ConnectionException.class)
|
||||
public void removeProxyException() {
|
||||
NettyProxyCenter.removeProxy(ProxyInterface.class, "localhost8894");
|
||||
}
|
||||
|
||||
@Test(expected = RuntimeException.class)
|
||||
public void getProxyTest() {
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", port.getPort());
|
||||
NettyClientPoolHandler handler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
ProxyClass localhost = NettyProxyCenter.getProxy(ProxyClass.class, address, handler);
|
||||
Assert.assertNotNull(localhost);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bindPipelineTest() throws IOException {
|
||||
// server
|
||||
Class<InstanceServerLoader> cls = InstanceServerLoader.class;
|
||||
ClassRegistry.put(cls.getName(), cls);
|
||||
ServerPort port = new TestServerPort();
|
||||
Instance instance = new DefaultInstance();
|
||||
NettyServerTakeHandler serverHandler = new NettyServerTakeHandler(instance);
|
||||
NettyServerConnection connection = new NettyServerConnection(serverHandler);
|
||||
RPCServer rpcServer = new RPCServer(connection, port);
|
||||
rpcServer.bind();
|
||||
while (!rpcServer.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
InetSocketAddress address = InetSocketAddress.createUnresolved("localhost", port.getPort());
|
||||
ChannelPoolHandler channelPoolHandler = new NettyClientPoolHandler(new NettyClientTakeHandler());
|
||||
NettyClientConnection clientConnection = new NettyClientConnection(address, channelPoolHandler);
|
||||
RPCClient rpcClient = new RPCClient(clientConnection);
|
||||
|
||||
InstanceServerLoader loader = NettyProxyCenter.createProxy(rpcClient, cls, address);
|
||||
String name = loader.getName();
|
||||
Assert.assertEquals("name", name);
|
||||
rpcClient.close();
|
||||
rpcServer.close();
|
||||
}
|
||||
|
||||
interface ProxyInterface {
|
||||
|
||||
void hello();
|
||||
}
|
||||
|
||||
static class ProxyClass {
|
||||
|
||||
}
|
||||
|
||||
static class TestServerPort implements ServerPort {
|
||||
|
||||
int port = RandomPort.getSafeRandomPort();
|
||||
|
||||
@Override
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,56 +0,0 @@
|
||||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright ownership.
|
||||
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
* (the "License"); you may not use this file except in compliance with
|
||||
* the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package cn.hippo4j.rpc.support;
|
||||
|
||||
import cn.hippo4j.common.toolkit.ThreadUtil;
|
||||
import cn.hippo4j.rpc.client.RandomPort;
|
||||
import cn.hippo4j.rpc.discovery.InstanceServerLoader;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class NettyServerSupportTest {
|
||||
|
||||
@Test
|
||||
public void bind() throws IOException {
|
||||
NettyServerSupport support = new NettyServerSupport(RandomPort::getSafeRandomPort, InstanceServerLoader.class);
|
||||
support.bind();
|
||||
while (!support.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
Assert.assertTrue(support.isActive());
|
||||
support.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bindTest() throws IOException {
|
||||
List<Class<?>> classes = new ArrayList<>();
|
||||
classes.add(InstanceServerLoader.class);
|
||||
NettyServerSupport support = new NettyServerSupport(RandomPort::getSafeRandomPort, classes);
|
||||
support.bind();
|
||||
while (!support.isActive()) {
|
||||
ThreadUtil.sleep(100L);
|
||||
}
|
||||
Assert.assertTrue(support.isActive());
|
||||
support.close();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,18 +0,0 @@
|
||||
#
|
||||
# Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
# contributor license agreements. See the NOTICE file distributed with
|
||||
# this work for additional information regarding copyright ownership.
|
||||
# The ASF licenses this file to You under the Apache License, Version 2.0
|
||||
# (the "License"); you may not use this file except in compliance with
|
||||
# the License. You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
#
|
||||
|
||||
cn.hippo4j.rpc.discovery.InstanceServerLoaderImpl
|
||||
Loading…
Reference in new issue