ehlxr/did
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

use java8 functional

Browse Source
This commit is contained in:
ehlxr 2021-02-08 17:15:36 +08:00
parent e929cd180f
commit 169bf6f290
20 changed files with 536 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
did-common/src/main/java/io/github/ehlxr/did/common/NettyUtil.java
View File

@ -5,6 +5,7 @@ import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.net.SocketAddress;
/**
@ -39,4 +40,19 @@ public class NettyUtil {
logger.info("closeChannel: close the connection to remote address[{}] result: {}",
addrRemote, future.isSuccess()));
}
public static byte[] toBytes(Object obj) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(obj);
oos.flush();
return bos.toByteArray();
}
public static Object toObject(byte[] bts) throws IOException, ClassNotFoundException {
ByteArrayInputStream bis = new ByteArrayInputStream(bts);
ObjectInputStream ois = new ObjectInputStream(bis);
return ois.readObject();
}
}

4
did-common/src/main/java/io/github/ehlxr/did/common/SdkProto.java
View File

@ -1,12 +1,14 @@
package io.github.ehlxr.did.common;
import java.io.Serializable;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author ehlxr
*/
public class SdkProto {
public class SdkProto implements Serializable {
private static final AtomicInteger REQUEST_ID = new AtomicInteger(0);
private static final long serialVersionUID = 8986093068588745760L;
/**
* 请求的ID

377
did-common/src/main/java/io/github/ehlxr/did/common/Try.java Normal file
View File

@ -0,0 +1,377 @@
/*
* The MIT License (MIT)
*
* Copyright © 2020 xrv <xrg@live.com>
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*/
package io.github.ehlxr.did.common;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Consumer;
/**
* 异常处理，简化 try catch
*
* @author ehlxr
* @since 2020-12-03 10:37.
*/
public interface Try {
/**
* 构建消费型（有入参，无返回）Tryable 对象
*
* @param consumer {@link ThrowableConsumer} 类型函数式接口
* @param <P> 入参类型
* @return {@link TryConsumer}
*/
static <P> TryConsumer<P> of(ThrowableConsumer<? super P> consumer) {
return new TryConsumer<>(consumer);
}
/**
* 构建供给型（无入参，有返回）Tryable 对象
*
* @param supplier {@link ThrowableSupplier} 类型函数式接口
* @param <R> 返回值类型
* @return {@link TrySupplier}
*/
static <R> TrySupplier<R> of(ThrowableSupplier<? extends R> supplier) {
return new TrySupplier<>(supplier);
}
/**
* 构建功能型（有入参，有返回）Tryable 对象
*
* @param function {@link ThrowableFunction} 类型函数式接口
* @param <P> 入参类型
* @param <R> 返回值类型
* @return {@link TryFunction}
*/
static <P, R> TryFunction<P, R> of(ThrowableFunction<? super P, ? extends R> function) {
return new TryFunction<>(function);
}
/**
* 构建运行型（无入参，无返回）Tryable 对象
*
* @param runnable {@link ThrowableRunnable} 类型函数式接口
* @return {@link TryRunnable}
*/
static TryRunnable of(ThrowableRunnable runnable) {
return new TryRunnable(runnable);
}
@FunctionalInterface
interface ThrowableConsumer<P> {
/**
* Performs this operation on the given argument.
*
* @param p the input argument
* @throws Throwable throwable
*/
void accept(P p) throws Throwable;
}
@FunctionalInterface
interface ThrowableSupplier<R> {
/**
* Gets a result.
*
* @return a result
* @throws Throwable throwable
*/
R get() throws Throwable;
}
@FunctionalInterface
interface ThrowableRunnable {
/**
* Performs this operation
*
* @throws Throwable throwable
*/
void run() throws Throwable;
}
@FunctionalInterface
interface ThrowableFunction<P, R> {
/**
* Applies this function to the given argument.
*
* @param p the function argument
* @return the function result
* @throws Throwable throwable
*/
R apply(P p) throws Throwable;
}
abstract class Tryable<C> {
Consumer<? super Throwable> throwConsumer;
ThrowableRunnable finallyRunnable;
ThrowableConsumer<? super Object> finallyConsumer;
Consumer<? super Throwable> finallyThrowConsumer;
C c;
/**
* 处理 finally
*/
protected void doFinally() {
Optional.ofNullable(finallyRunnable).ifPresent(r -> {
try {
r.run();
} catch (final Throwable e) {
Optional.ofNullable(finallyThrowConsumer).ifPresent(tc -> tc.accept(e));
}
});
}
/**
* 处理带参数类型 finally
*
* @param p 入参
* @param <P> 入参类型
*/
protected <P> void doFinally(P p) {
if (Objects.nonNull(finallyConsumer)) {
try {
finallyConsumer.accept(p);
} catch (final Throwable e) {
Optional.ofNullable(finallyThrowConsumer).ifPresent(tc -> tc.accept(e));
}
} else {
doFinally();
}
}
/**
* 如果有异常，调用自定义异常处理表达式
*
* @param throwableConsumer 自定义异常处理 lambda 表达式
* @return {@link C}
*/
public C trap(Consumer<? super Throwable> throwableConsumer) {
Objects.requireNonNull(throwableConsumer, "No throwableConsumer present");
this.throwConsumer = throwableConsumer;
return c;
}
/**
* 自定义 finally 处理表达式
* <p>
* 注意：如果类型为 {@link TryConsumer}、{@link TryFunction} 并且已经调用 {@link #andFinally(ThrowableConsumer)} 则忽略
*
* @param finallyRunnable finally 处理 lambda 表达式
* @return {@link C}
*/
public C andFinally(ThrowableRunnable finallyRunnable) {
Objects.requireNonNull(finallyRunnable, "No finallyRunnable present");
this.finallyRunnable = finallyRunnable;
return c;
}
/**
* 自定义 finally 处理表达式
* <p>
* 注意：只会对 {@link TryConsumer}、{@link TryFunction} 类型对象起作用
*
* @param finallyConsumer finally 处理 lambda 表达式
* @return {@link C}
*/
public C andFinally(ThrowableConsumer<? super Object> finallyConsumer) {
Objects.requireNonNull(finallyConsumer, "No finallyConsumer present");
this.finallyConsumer = finallyConsumer;
return c;
}
/**
* 如果 finally 有异常，调用自定义异常处理表达式
*
* @param finallyThrowableConsumer 自定义异常处理 lambda 表达式
* @return {@link C}
*/
public C finallyTrap(Consumer<? super Throwable> finallyThrowableConsumer) {
Objects.requireNonNull(finallyThrowableConsumer, "No finallyThrowableConsumer present");
this.finallyThrowConsumer = finallyThrowableConsumer;
return c;
}
}
class TryRunnable extends Tryable<TryRunnable> {
private final ThrowableRunnable runnable;
protected TryRunnable(ThrowableRunnable runnable) {
Objects.requireNonNull(runnable, "No runnable present");
this.runnable = runnable;
super.c = this;
}
/**
* 计算结果
*/
public void run() {
try {
runnable.run();
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
} finally {
doFinally();
}
}
}
class TrySupplier<R> extends Tryable<TrySupplier<R>> {
private final ThrowableSupplier<? extends R> supplier;
protected TrySupplier(ThrowableSupplier<? extends R> supplier) {
Objects.requireNonNull(supplier, "No supplier present");
this.supplier = supplier;
super.c = this;
}
/**
* 如果有异常返回默认值，否则返回计算结果
*
* @param r 指定默认值
* @return 实际值或默认值
*/
public R get(R r) {
try {
return supplier.get();
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
return r;
} finally {
doFinally();
}
}
/**
* 如果有异常返回 null，否则返回计算结果
*
* @return 实际值或 null
*/
public R get() {
try {
return supplier.get();
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
return null;
} finally {
doFinally();
}
}
}
class TryConsumer<P> extends Tryable<TryConsumer<P>> {
private final ThrowableConsumer<? super P> consumer;
protected TryConsumer(ThrowableConsumer<? super P> consumer) {
Objects.requireNonNull(consumer, "No consumer present");
this.consumer = consumer;
super.c = this;
}
/**
* 计算结果
*
* @param p 要计算的入参
*/
public void accept(P p) {
try {
Objects.requireNonNull(p, "No accept param present");
consumer.accept(p);
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
} finally {
// doFinally();
doFinally(p);
}
}
}
class TryFunction<P, R> extends Tryable<TryFunction<P, R>> {
private final ThrowableFunction<? super P, ? extends R> function;
private P p;
protected TryFunction(ThrowableFunction<? super P, ? extends R> function) {
Objects.requireNonNull(function, "No function present");
this.function = function;
super.c = this;
}
/**
* 传入要计算的入参
*
* @param p 要计算的入参
* @return {@link TryFunction}
*/
public TryFunction<P, R> apply(P p) {
Objects.requireNonNull(p, "Apply param should not null");
this.p = p;
return this;
}
/**
* 如果有异常返回默认值，否则返回计算结果
*
* @param r 指定默认值
* @return 实际值或默认值
*/
public R get(R r) {
try {
Objects.requireNonNull(function, "No apply param present");
return function.apply(p);
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
return r;
} finally {
// doFinally();
doFinally(p);
}
}
/**
* 如果有异常返回 null，否则返回计算结果
*
* @return 实际值或 null
*/
public R get() {
try {
Objects.requireNonNull(p, "No apply param present");
return function.apply(p);
} catch (final Throwable e) {
Optional.ofNullable(throwConsumer).ifPresent(tc -> tc.accept(e));
return null;
} finally {
// doFinally();
doFinally(p);
}
}
}
}

58
did-sdk/src/main/java/io/github/ehlxr/did/client/AbstractClient.java
View File

@ -1,9 +1,6 @@
package io.github.ehlxr.did.client;
import io.github.ehlxr.did.common.Constants;
import io.github.ehlxr.did.common.NettyUtil;
import io.github.ehlxr.did.common.Result;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@ -14,6 +11,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Optional;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
@ -59,8 +57,8 @@ public abstract class AbstractClient implements Client {
}
});
bootstrap = new Bootstrap();
bootstrap.group(workGroup)
bootstrap = new Bootstrap()
.group(workGroup)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
// .option(ChannelOption.TCP_NODELAY, true)
// .option(ChannelOption.SO_KEEPALIVE, true)
@ -70,13 +68,11 @@ public abstract class AbstractClient implements Client {
@Override
public void shutdown() {
logger.info("SDK Client shutdowning......");
try {
if (workGroup != null) {
workGroup.shutdownGracefully().sync();
}
} catch (Exception e) {
logger.error("Client EventLoopGroup shutdown error.", e);
}
Optional.ofNullable(workGroup).ifPresent(wg ->
Try.<NioEventLoopGroup>of(w -> w.shutdownGracefully().sync())
.trap(e -> logger.error("Client EventLoopGroup shutdown error.", e))
.accept(wg));
logger.info("SDK Client shutdown finish!");
}
@ -87,9 +83,12 @@ public abstract class AbstractClient implements Client {
if (channel.isOpen() && channel.isActive()) {
final SdkProto sdkProto = new SdkProto();
final int rqid = sdkProto.getRqid();
try {
return Try.of(() -> {
final ResponseFuture responseFuture = new ResponseFuture(timeoutMillis, null, null);
REPONSE_MAP.put(rqid, responseFuture);
logger.debug("write {} to channel", sdkProto);
channel.writeAndFlush(sdkProto).addListener((ChannelFutureListener) channelFuture -> {
if (channelFuture.isSuccess()) {
//发送成功后立即跳出
@ -101,18 +100,20 @@ public abstract class AbstractClient implements Client {
responseFuture.setCause(channelFuture.cause());
logger.error("send a request command to channel <{}> failed.", NettyUtil.parseRemoteAddr(channel));
});
// 阻塞等待响应
SdkProto proto = responseFuture.waitResponse(timeoutMillis);
if (null == proto) {
return Result.fail("get result fail, addr is " + NettyUtil.parseRemoteAddr(channel) + responseFuture.getCause());
String msg = String.format("get result from addr %s failed, cause by %s",
NettyUtil.parseRemoteAddr(channel), responseFuture.getCause());
logger.error(msg);
return Result.<SdkProto>fail(msg);
}
return Result.success(proto);
} catch (Exception e) {
logger.error("invokeSync fail, addr is " + NettyUtil.parseRemoteAddr(channel), e);
return Result.fail("invokeSync fail, addr is " + NettyUtil.parseRemoteAddr(channel) + e.getMessage());
} finally {
REPONSE_MAP.remove(rqid);
}
}).trap(e -> logger.error("sync invoke {} failed.", NettyUtil.parseRemoteAddr(channel), e))
.andFinally(() -> REPONSE_MAP.remove(rqid))
.get(Result.fail("failed to sync invoke " + NettyUtil.parseRemoteAddr(channel)));
} else {
NettyUtil.closeChannel(channel);
return Result.fail("channel " + NettyUtil.parseRemoteAddr(channel) + "is not active!");
@ -128,7 +129,9 @@ public abstract class AbstractClient implements Client {
if (asyncSemaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS)) {
final ResponseFuture responseFuture = new ResponseFuture(timeoutMillis, invokeCallback, asyncSemaphore);
REPONSE_MAP.put(rqid, responseFuture);
try {
Try.of(() -> {
logger.debug("write {} to channel", sdkProto);
channelFuture.channel().writeAndFlush(sdkProto).addListener(channelFuture -> {
if (channelFuture.isSuccess()) {
return;
@ -141,18 +144,13 @@ public abstract class AbstractClient implements Client {
REPONSE_MAP.remove(rqid);
responseFuture.setCause(channelFuture.cause());
responseFuture.putResponse(null);
responseFuture.executeInvokeCallback();
responseFuture.release();
});
} catch (Exception e) {
}).trap(e -> {
responseFuture.release();
String msg = String.format("send a request to channel <%s> Exception",
NettyUtil.parseRemoteAddr(channel));
logger.error(msg, e);
throw new Exception(msg, e);
}
logger.error("send a request to channel <{}> Exception", NettyUtil.parseRemoteAddr(channel), e);
}).run();
} else {
String msg = String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread " +
"nums: %d semaphoreAsyncValue: %d",

2
did-sdk/src/main/java/io/github/ehlxr/did/client/Client.java
View File

@ -28,7 +28,6 @@ public interface Client {
*
* @param timeoutMillis 超时时间
* @return {@link SdkProto}
* @throws Exception 调用异常
*/
Result<SdkProto> invokeSync(long timeoutMillis);
@ -46,7 +45,6 @@ public interface Client {
*
* @param timeoutMillis 超时时间
* @return id
* @throws Exception 调用异常
*/
default Result<SdkProto> invoke(long timeoutMillis) {
return invokeSync(timeoutMillis);

9
did-sdk/src/main/java/io/github/ehlxr/did/client/ResponseFuture.java
View File

@ -1,6 +1,7 @@
package io.github.ehlxr.did.client;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.Try;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
@ -46,8 +47,12 @@ public class ResponseFuture {
}
}
public SdkProto waitResponse(final long timeoutMillis) throws InterruptedException {
this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
public SdkProto waitResponse(final long timeoutMillis) {
Try.of(() -> {
if (!this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS)) {
setCause(new RuntimeException("timeout after wait " + timeoutMillis));
}
}).trap(this::setCause).run();
return this.sdkProto;
}

14
did-sdk/src/main/java/io/github/ehlxr/did/client/SdkClient.java
View File

@ -4,6 +4,7 @@ import io.github.ehlxr.did.client.handler.SdkClientDecoder;
import io.github.ehlxr.did.client.handler.SdkClientEncoder;
import io.github.ehlxr.did.client.handler.SdkClientHandler;
import io.github.ehlxr.did.common.Constants;
import io.github.ehlxr.did.common.Try;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
@ -50,13 +51,14 @@ public class SdkClient extends AbstractClient {
bootstrap.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) {
socketChannel.pipeline().addLast("SdkServerDecoder", new SdkClientDecoder(12))
.addLast("SdkServerEncoder", new SdkClientEncoder())
.addLast("SdkClientHandler", new SdkClientHandler());
socketChannel.pipeline()
.addLast(new SdkClientDecoder())
.addLast(new SdkClientEncoder())
.addLast(new SdkClientHandler());
}
});
try {
Try.of(() -> {
channelFuture = bootstrap.connect(host, port)
.sync()
.channel()
@ -66,10 +68,10 @@ public class SdkClient extends AbstractClient {
InetSocketAddress address = (InetSocketAddress) channelFuture.channel().remoteAddress();
logger.info("SdkClient start success, host is {}, port is {}", address.getHostName(), address.getPort());
} catch (InterruptedException e) {
}).trap(e -> {
logger.error("SdkClient start error", e);
shutdown();
}
}).run();
}
public static final class SdkClientBuilder {

40
did-sdk/src/main/java/io/github/ehlxr/did/client/handler/SdkClientDecoder.java
View File

@ -25,49 +25,37 @@
package io.github.ehlxr.did.client.handler;
import io.github.ehlxr.did.common.NettyUtil;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.Try;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
/**
* @author ehlxr
* @since 2021-01-20 14:42.
*/
public class SdkClientDecoder extends FixedLengthFrameDecoder {
public class SdkClientDecoder extends ByteToMessageDecoder {
private final Logger logger = LoggerFactory.getLogger(SdkClientDecoder.class);
public SdkClientDecoder(int frameLength) {
super(frameLength);
}
@Override
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) {
ByteBuf buf = null;
try {
buf = (ByteBuf) super.decode(ctx, in);
if (buf == null) {
return null;
}
return new SdkProto(buf.readInt(), buf.readLong());
} catch (Exception e) {
logger.error("SdkClientDecoder decode exception, " + NettyUtil.parseRemoteAddr(ctx.channel()), e);
NettyUtil.closeChannel(ctx.channel());
} finally {
if (buf != null) {
buf.release();
}
}
return null;
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
Try.of(() -> {
byte[] bytes = new byte[in.readableBytes()];
in.readBytes(bytes);
out.add(NettyUtil.toObject(bytes));
}).trap(e -> logger.error("decode error", e)).run();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
Channel channel = ctx.channel();
logger.error("SdkServerDecoder channel [{}] error and will be closed", NettyUtil.parseRemoteAddr(channel), cause);
logger.error("SdkClientDecoder channel [{}] error and will be closed", NettyUtil.parseRemoteAddr(channel), cause);
NettyUtil.closeChannel(channel);
}
}
}

8
did-sdk/src/main/java/io/github/ehlxr/did/client/handler/SdkClientEncoder.java
View File

@ -27,6 +27,7 @@ package io.github.ehlxr.did.client.handler;
import io.github.ehlxr.did.client.SdkClient;
import io.github.ehlxr.did.common.NettyUtil;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.Try;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@ -41,11 +42,12 @@ import org.slf4j.LoggerFactory;
public class SdkClientEncoder extends MessageToByteEncoder<SdkProto> {
private final Logger logger = LoggerFactory.getLogger(SdkClient.class);
@Override
protected void encode(ChannelHandlerContext ctx, SdkProto sdkProto, ByteBuf out) {
out.writeInt(sdkProto.getRqid());
out.writeLong(sdkProto.getDid());
System.out.println("-------------");
Try.of(() -> {
out.writeBytes(NettyUtil.toBytes(sdkProto));
}).trap(e -> logger.error("encode error", e)).run();
}
@Override

2
did-sdk/src/main/java/io/github/ehlxr/did/client/handler/SdkClientHandler.java
View File

@ -42,6 +42,8 @@ public class SdkClientHandler extends SimpleChannelInboundHandler<SdkProto> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, SdkProto sdkProto) {
logger.debug("sdk client handler receive sdkProto {}", sdkProto);
final int rqid = sdkProto.getRqid();
final ResponseFuture responseFuture = Client.REPONSE_MAP.get(rqid);
if (responseFuture != null) {

16
did-sdk/src/main/resources/logback.xml Normal file
View File

@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%date [%thread] %-5level %logger{35}:%line - %msg%n</pattern>
<charset>UTF-8</charset>
</encoder>
</appender>
<root level="DEBBUG">
<appender-ref ref="STDOUT"/>
</root>
<logger name="io.netty" level="OFF"/>
</configuration>

36
did-sdk/src/test/java/io/github/ehlxr/did/DidSdkTest.java
View File

@ -1,14 +1,11 @@
package io.github.ehlxr.did;
import io.github.ehlxr.did.client.SdkClient;
import io.github.ehlxr.did.common.Result;
import io.github.ehlxr.did.common.SdkProto;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
/**
* @author ehlxr
@ -20,7 +17,7 @@ public class DidSdkTest {
@Before
public void init() {
// client = new SdkClient("127.0.0.1", 16831, 5000);
client = SdkClient.newBuilder().build();
client = SdkClient.newBuilder().timeoutMillis(5000).build();
client.start();
}
@ -32,27 +29,26 @@ public class DidSdkTest {
@Test
public void didSdkTest() throws Exception {
// 测试同步请求
for (int i = 0; i < NUM; i++) {
Result<SdkProto> resultProto = client.invokeSync();
System.out.println(resultProto);
}
System.out.println("invokeync test finish");
IntStream.range(0, NUM).parallel().forEach(i -> System.out.println(client.invokeSync()));
// System.out.println("invokeync test finish");
// 测试异步请求
final CountDownLatch countDownLatch = new CountDownLatch(NUM);
for (int i = 0; i < NUM; i++) {
client.invokeAsync(responseFuture -> {
countDownLatch.countDown();
System.out.println(responseFuture.getSdkProto());
});
}
countDownLatch.await(10, TimeUnit.SECONDS);
System.out.println("invokeAsync test finish");
// final CountDownLatch countDownLatch = new CountDownLatch(NUM);
// IntStream.range(0, NUM).forEach(i ->
// Try.of(() -> client.invokeAsync(responseFuture -> {
// System.out.println(responseFuture.getSdkProto());
// countDownLatch.countDown();
// })).trap(Throwable::printStackTrace).run());
//
// //noinspection ResultOfMethodCallIgnored
// countDownLatch.await(10, TimeUnit.SECONDS);
// System.out.println("invokeAsync test finish");
}
@Test
public void testInvoke() {
System.out.println(client.invoke());
// System.out.println(client.invoke());
client.setTimeoutMillis(3000);
System.out.println(client.invoke());

12
did-server/src/main/java/io/github/ehlxr/did/server/BaseServer.java
View File

@ -1,5 +1,6 @@
package io.github.ehlxr.did.server;
import io.github.ehlxr.did.common.Try;
import io.github.ehlxr.did.core.SnowFlake;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
@ -52,9 +53,8 @@ public abstract class BaseServer implements Server {
}
});
serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workGroup)
serverBootstrap = new ServerBootstrap()
.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
// .option(ChannelOption.SO_KEEPALIVE, true)
// .option(ChannelOption.TCP_NODELAY, true)
@ -64,15 +64,13 @@ public abstract class BaseServer implements Server {
@Override
public void shutdown() {
try {
Try.of(() -> {
// 同步阻塞 shutdownGracefully 完成
if (defLoopGroup != null) {
defLoopGroup.shutdownGracefully().sync();
}
bossGroup.shutdownGracefully().sync();
workGroup.shutdownGracefully().sync();
} catch (Exception e) {
logger.error("Server EventLoopGroup shutdown error.", e);
}
}).trap(e -> logger.error("Server EventLoopGroup shutdown error.", e)).run();
}
}

7
did-server/src/main/java/io/github/ehlxr/did/server/http/HttpServer.java
View File

@ -1,6 +1,7 @@
package io.github.ehlxr.did.server.http;
import io.github.ehlxr.did.common.Constants;
import io.github.ehlxr.did.common.Try;
import io.github.ehlxr.did.core.SnowFlake;
import io.github.ehlxr.did.server.BaseServer;
import io.netty.channel.ChannelInitializer;
@ -46,7 +47,7 @@ public class HttpServer extends BaseServer {
@Override
public void start() {
try {
Try.of(() -> {
init();
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@ -63,9 +64,7 @@ public class HttpServer extends BaseServer {
channelFuture = serverBootstrap.bind(port).sync();
logger.info("HttpServer start success, port is:{}", port);
} catch (InterruptedException e) {
logger.error("HttpServer start fail,", e);
}
}).trap(e -> logger.error("HttpServer start fail,", e)).run();
}
public static final class HttpServerBuilder {

20
did-server/src/main/java/io/github/ehlxr/did/server/http/HttpServerHandler.java
View File

@ -33,6 +33,8 @@ public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpReque
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
logger.debug("http server handler receive request {}", request);
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
Result<?> result;
@ -48,23 +50,14 @@ public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpReque
}
if (semaphore.tryAcquire(Constants.ACQUIRE_TIMEOUTMILLIS, TimeUnit.MILLISECONDS)) {
try {
long id = snowFlake.nextId();
logger.info("HttpServerHandler id is: {}", id);
long id = snowFlake.nextId();
status = HttpResponseStatus.OK;
result = Result.success(SdkProto.newBuilder().did(id).build());
} catch (Exception e) {
semaphore.release();
logger.error("HttpServerHandler error", e);
status = HttpResponseStatus.INTERNAL_SERVER_ERROR;
result = Result.fail(status.code(), status.reasonPhrase());
}
status = HttpResponseStatus.OK;
result = Result.success(SdkProto.newBuilder().did(id).build());
} else {
String info = String.format("HttpServerHandler tryAcquire semaphore timeout, %dms, waiting thread " + "nums: %d availablePermit: %d",
Constants.ACQUIRE_TIMEOUTMILLIS, this.semaphore.getQueueLength(), this.semaphore.availablePermits());
logger.warn(info);
logger.error(info);
status = HttpResponseStatus.SERVICE_UNAVAILABLE;
result = Result.fail(status.code(), info);
@ -73,6 +66,7 @@ public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpReque
response.setStatus(status)
.content().writeBytes(result.toString().getBytes());
logger.debug("http server handler write response {} restul {} to channel", status, result);
ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
}

10
did-server/src/main/java/io/github/ehlxr/did/server/sdk/SdkServer.java
View File

@ -1,6 +1,7 @@
package io.github.ehlxr.did.server.sdk;
import io.github.ehlxr.did.common.Constants;
import io.github.ehlxr.did.common.Try;
import io.github.ehlxr.did.core.SnowFlake;
import io.github.ehlxr.did.server.BaseServer;
import io.netty.channel.ChannelInitializer;
@ -41,14 +42,15 @@ public class SdkServer extends BaseServer {
@Override
public void start() {
try {
Try.of(() -> {
init();
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(defLoopGroup,
new SdkServerDecoder(12),
new SdkServerDecoder(),
// new SdkServerDecoder(12),
new SdkServerEncoder(),
new SdkServerHandler(snowFlake)
);
@ -57,9 +59,7 @@ public class SdkServer extends BaseServer {
channelFuture = serverBootstrap.bind(port).sync();
logger.info("SdkServer start success, port is:{}", port);
} catch (InterruptedException e) {
logger.error("SdkServer start fail,", e);
}
}).trap(e -> logger.error("SdkServer start fail,", e)).run();
}
public static final class SdkServerBuilder {

40
did-server/src/main/java/io/github/ehlxr/did/server/sdk/SdkServerDecoder.java
View File

@ -1,42 +1,30 @@
package io.github.ehlxr.did.server.sdk;
import io.github.ehlxr.did.common.NettyUtil;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.Try;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
/**
* @author ehlxr
*/
public class SdkServerDecoder extends FixedLengthFrameDecoder {
private static final Logger logger = LoggerFactory.getLogger(SdkServerDecoder.class);
SdkServerDecoder(int frameLength) {
super(frameLength);
}
public class SdkServerDecoder extends ByteToMessageDecoder {
private final Logger logger = LoggerFactory.getLogger(SdkServerDecoder.class);
@Override
protected Object decode(ChannelHandlerContext ctx, ByteBuf in) {
ByteBuf buf = null;
try {
buf = (ByteBuf) super.decode(ctx, in);
if (buf == null) {
return null;
}
return new SdkProto(buf.readInt(), buf.readLong());
} catch (Exception e) {
logger.error("decode exception, " + NettyUtil.parseRemoteAddr(ctx.channel()), e);
NettyUtil.closeChannel(ctx.channel());
} finally {
if (buf != null) {
buf.release();
}
}
return null;
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
Try.of(() -> {
byte[] bytes = new byte[in.readableBytes()];
in.readBytes(bytes);
out.add(NettyUtil.toObject(bytes));
}).trap(e -> logger.error("decode error", e)).run();
}
@Override
@ -45,4 +33,4 @@ public class SdkServerDecoder extends FixedLengthFrameDecoder {
logger.error("SdkServerDecoder channel [{}] error and will be closed", NettyUtil.parseRemoteAddr(channel), cause);
NettyUtil.closeChannel(channel);
}
}
}

7
did-server/src/main/java/io/github/ehlxr/did/server/sdk/SdkServerEncoder.java
View File

@ -2,6 +2,7 @@ package io.github.ehlxr.did.server.sdk;
import io.github.ehlxr.did.common.NettyUtil;
import io.github.ehlxr.did.common.SdkProto;
import io.github.ehlxr.did.common.Try;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
@ -15,11 +16,11 @@ import org.slf4j.LoggerFactory;
public class SdkServerEncoder extends MessageToByteEncoder<SdkProto> {
private static final Logger logger = LoggerFactory.getLogger(SdkServerEncoder.class);
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, SdkProto sdkProto, ByteBuf out) {
out.writeInt(sdkProto.getRqid());
out.writeLong(sdkProto.getDid());
Try.of(() -> {
out.writeBytes(NettyUtil.toBytes(sdkProto));
}).trap(e -> logger.error("encode error", e)).run();
}
@Override

3
did-server/src/main/java/io/github/ehlxr/did/server/sdk/SdkServerHandler.java
View File

@ -33,6 +33,8 @@ public class SdkServerHandler extends SimpleChannelInboundHandler<SdkProto> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, SdkProto sdkProto) throws Exception {
logger.debug("sdk server handler receive sdkProto {}", sdkProto);
if (semaphore.tryAcquire(Constants.ACQUIRE_TIMEOUTMILLIS, TimeUnit.MILLISECONDS)) {
sdkProto.setDid(snowFlake.nextId());
@ -42,6 +44,7 @@ public class SdkServerHandler extends SimpleChannelInboundHandler<SdkProto> {
Constants.ACQUIRE_TIMEOUTMILLIS, this.semaphore.getQueueLength(), this.semaphore.availablePermits());
}
logger.debug("sdk server handler write sdkProto {} to channel", sdkProto);
ctx.channel().
writeAndFlush(sdkProto).
addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

4
did-server/src/main/resources/logback.xml
View File

@ -36,10 +36,12 @@
</encoder>
</appender>
<root level="INFO">
<root level="DEBUUG">
<appender-ref ref="STDOUT"/>
<appender-ref ref="ERROR"/>
<appender-ref ref="NORMAL"/>
</root>
<logger name="io.netty" level="OFF"/>
</configuration>