-
Notifications
You must be signed in to change notification settings - Fork 154
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore: update multiple files without content changes
- Loading branch information
1 parent
2e1a60b
commit 18d9e33
Showing
9 changed files
with
210 additions
and
35 deletions.
There are no files selected for viewing
15 changes: 15 additions & 0 deletions
15
jcommon/api/src/main/java/com/xiaomi/data/push/uds/processor/StreamCallback.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package com.xiaomi.data.push.uds.processor; | ||
|
|
||
| /** | ||
| * @author [email protected] | ||
| * @date 2024/11/7 11:56 | ||
| */ | ||
| public interface StreamCallback { | ||
|
|
||
| void onContent(String content); | ||
|
|
||
| void onComplete(); | ||
|
|
||
| void onError(Throwable error); | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
15 changes: 15 additions & 0 deletions
15
jcommon/rcurve/src/main/java/com/xiaomi/data/push/uds/handler/ClientStreamCallback.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package com.xiaomi.data.push.uds.handler; | ||
|
|
||
| /** | ||
| * @author [email protected] | ||
| * @date 2024/11/7 10:35 | ||
| */ | ||
| public interface ClientStreamCallback { | ||
|
|
||
| void onContent(String content); | ||
|
|
||
| void onComplete(); | ||
|
|
||
| void onError(Throwable error); | ||
|
|
||
| } |
18 changes: 18 additions & 0 deletions
18
jcommon/rcurve/src/main/java/com/xiaomi/data/push/uds/handler/MessageTypes.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package com.xiaomi.data.push.uds.handler; | ||
|
|
||
| /** | ||
| * @author [email protected] | ||
| * @date 2024/11/6 17:41 | ||
| */ | ||
| public class MessageTypes { | ||
|
|
||
| public static final String TYPE_KEY = "messageType"; | ||
| public static final String TYPE_NORMAL = "normal"; | ||
| public static final String TYPE_OPENAI = "openai"; | ||
| public static final String STREAM_ID_KEY = "streamId"; | ||
| public static final String PROMPT_KEY = "prompt"; | ||
| public static final String CONTENT_KEY = "content"; | ||
| public static final String STATUS_KEY = "status"; | ||
|
|
||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,6 +21,7 @@ | |
| import com.xiaomi.data.push.uds.UdsServer; | ||
| import com.xiaomi.data.push.uds.context.UdsServerContext; | ||
| import com.xiaomi.data.push.uds.po.UdsCommand; | ||
| import com.xiaomi.data.push.uds.processor.StreamCallback; | ||
| import com.xiaomi.data.push.uds.processor.UdsProcessor; | ||
| import io.netty.buffer.ByteBuf; | ||
| import io.netty.channel.ChannelHandlerContext; | ||
|
|
@@ -32,7 +33,9 @@ | |
|
|
||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.UUID; | ||
| import java.util.concurrent.*; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| /** | ||
| * @author [email protected] | ||
|
|
@@ -45,33 +48,30 @@ public class UdsServerHandler extends ChannelInboundHandlerAdapter { | |
|
|
||
| private Map<String, Pair<UdsProcessor, ExecutorService>> m; | ||
|
|
||
|
|
||
| public UdsServerHandler(ConcurrentHashMap<String, Pair<UdsProcessor, ExecutorService>> processorMap) { | ||
| this.m = processorMap; | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public void channelRead(ChannelHandlerContext ctx, Object _msg) { | ||
| try { | ||
| ByteBuf msg = (ByteBuf) _msg; | ||
| UdsCommand command = new UdsCommand(); | ||
| command.decode(msg); | ||
| log.debug("server receive:id:{}:{}:{}:{}:{}",command.getId(), command.isRequest(), command.getApp(), command.getCmd(), command.getSerializeType()); | ||
| log.debug("server receive:id:{}:{}:{}:{}:{}", command.getId(), command.isRequest(), command.getApp(), command.getCmd(), command.getSerializeType()); | ||
| if (command.isRequest()) { | ||
| command.setChannel(ctx.channel()); | ||
| Pair<UdsProcessor, ExecutorService> pair = this.m.get(command.getCmd()); | ||
| if (null != pair) { | ||
| UdsProcessor<UdsCommand, UdsCommand> processor = pair.getKey(); | ||
| pair.getValue().submit(() -> { | ||
| log.debug("server received:{}", command.getId()); | ||
| UdsCommand res = processor.processRequest(command); | ||
| if (null != res) { | ||
| Send.send(ctx.channel(), res); | ||
| } | ||
| }); | ||
| // 判断是否为流式处理 | ||
| if (processor.isStreamProcessor()) { | ||
| handleStreamRequest(ctx, command, processor); | ||
| } else { | ||
| handleNormalRequest(pair.getValue(), ctx, command, processor); | ||
| } | ||
| } else { | ||
| log.warn("processor is null cmd:{},id:{}", command.getCmd(),command.getId()); | ||
| log.warn("processor is null cmd:{},id:{}", command.getCmd(), command.getId()); | ||
| } | ||
| } else { | ||
| Optional.ofNullable(UdsServer.reqMap.get(command.getId())).ifPresent(f -> f.complete(command)); | ||
|
|
@@ -81,6 +81,74 @@ public void channelRead(ChannelHandlerContext ctx, Object _msg) { | |
| } | ||
| } | ||
|
|
||
| private void handleNormalRequest(ExecutorService pool, ChannelHandlerContext ctx, UdsCommand command, UdsProcessor<UdsCommand, UdsCommand> processor) { | ||
| pool.submit(() -> { | ||
| log.debug("server received:{}", command.getId()); | ||
| UdsCommand res = processor.processRequest(command); | ||
| if (null != res) { | ||
| Send.send(ctx.channel(), res); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
|
|
||
| private void handleStreamRequest(ChannelHandlerContext ctx, UdsCommand command, | ||
| UdsProcessor<UdsCommand, UdsCommand> processor) { | ||
|
|
||
| String streamId = command.getAttachments().getOrDefault( | ||
| MessageTypes.STREAM_ID_KEY, | ||
| UUID.randomUUID().toString() | ||
| ); | ||
|
|
||
| StreamCallback callback = new StreamCallback() { | ||
| @Override | ||
| public void onContent(String content) { | ||
| sendStreamContent(ctx, command, streamId, content); | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| sendCompleteResponse(ctx, command, streamId); | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable error) { | ||
| sendErrorResponse(ctx, command, error.getMessage()); | ||
| } | ||
| }; | ||
|
|
||
| // 执行流式处理 | ||
| processor.processStream(command, callback); | ||
| } | ||
|
|
||
|
|
||
| private void sendErrorResponse(ChannelHandlerContext ctx, UdsCommand command, String error) { | ||
| UdsCommand response = UdsCommand.createResponse(command); | ||
| response.setCode(-1); | ||
| response.setMessage(error); | ||
| Send.send(ctx.channel(), response); | ||
| } | ||
|
|
||
|
|
||
| private void sendCompleteResponse(ChannelHandlerContext ctx, UdsCommand request, String streamId) { | ||
| UdsCommand response = UdsCommand.createResponse(request); | ||
| Map<String, String> attachments = response.getAttachments(); | ||
| attachments.put(MessageTypes.TYPE_KEY, MessageTypes.TYPE_OPENAI); | ||
| attachments.put(MessageTypes.STREAM_ID_KEY, streamId); | ||
| attachments.put(MessageTypes.STATUS_KEY, "complete"); | ||
| Send.send(ctx.channel(), response); | ||
| } | ||
|
|
||
|
|
||
| private void sendStreamContent(ChannelHandlerContext ctx, UdsCommand request, String streamId, String content) { | ||
| UdsCommand response = UdsCommand.createResponse(request); | ||
| Map<String, String> attachments = response.getAttachments(); | ||
| attachments.put(MessageTypes.TYPE_KEY, MessageTypes.TYPE_OPENAI); | ||
| attachments.put(MessageTypes.STREAM_ID_KEY, streamId); | ||
| attachments.put(MessageTypes.CONTENT_KEY, content); | ||
| Send.send(ctx.channel(), response); | ||
| } | ||
|
|
||
|
|
||
| @Override | ||
| public void channelActive(ChannelHandlerContext ctx) throws Exception { | ||
|
|
@@ -91,15 +159,15 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception { | |
| public void channelInactive(ChannelHandlerContext ctx) { | ||
| Attribute<String> attr = ctx.channel().attr(app); | ||
| String v = attr.get(); | ||
| log.error("server channelInactive:{},{},{}", app, v,ctx.channel().id()); | ||
| log.error("server channelInactive:{},{},{}", app, v, ctx.channel().id()); | ||
| if (null != v) { | ||
| UdsServerContext.ins().remove(v); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { | ||
| log.error("exceptionCaught,{}:{}",ctx.channel().id(), cause); | ||
| log.error("exceptionCaught,{}:{}", ctx.channel().id(), cause); | ||
| Attribute<String> attr = ctx.channel().attr(app); | ||
| String v = attr.get(); | ||
| if (null != v) { | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters