!2 新增云快充1.6支持&Grpc下行接口

Merge pull request !2 from 三丙/feature/GrpcDownlink
This commit is contained in:
三丙
2024-10-25 06:32:06 +00:00
committed by Gitee
71 changed files with 1462 additions and 324 deletions

View File

@@ -15,9 +15,9 @@
------------------------------
#### 当前支持的充电桩协议
| 协议名 | 版本号 |
|---|---|
| 云快充 | 1.5.0 |
| 协议名 | 版本号 |
|---|------------|
| 云快充 | 1.5.0、1.6.0 |
------------------------------
#### 充电桩协议文档

View File

@@ -35,6 +35,9 @@ RUN chmod a+x *.sh && mv start.sh /usr/bin
EXPOSE 8080 8080
ENV APP_LOG_LEVEL=INFO
ENV PROTOCOLS_LOG_LEVEL=INFO
CMD ["start.sh"]

View File

@@ -35,6 +35,8 @@ RUN chmod a+x *.sh && mv start.sh /usr/bin
EXPOSE 8081 8081
ENV PROTOCOLS_LOG_LEVEL=INFO
CMD ["start.sh"]

View File

@@ -12,7 +12,7 @@ export JAVA_APP_OPTS="-XX:+UseContainerSupport -XX:InitialRAMPercentage=10 -XX:M
-XX:HeapDumpPath=/var/log/sanbing/heapdump/ \
-XX:+UseTLAB -XX:+ResizeTLAB -XX:+PerfDisableSharedMem -XX:+UseCondCardMark \
-XX:+UseG1GC -XX:MaxGCPauseMillis=500 -XX:+UseStringDeduplication -XX:+ParallelRefProcEnabled -XX:MaxTenuringThreshold=10 \
-Xss512k -XX:MaxDirectMemorySize=128M -XX:G1ReservePercent=20 \
-Xss512k -XX:MaxDirectMemorySize=256M -XX:G1ReservePercent=20 \
-XX:-OmitStackTraceInFastThrow \
-Dlogging.config=/app/config/log4j2.xml"

View File

@@ -75,10 +75,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

View File

@@ -27,7 +27,7 @@ spring:
password: "${SPRING_DATASOURCE_PASSWORD:postgres}"
hikari:
leak-detection-threshold: "${SPRING_DATASOURCE_HIKARI_LEAK_DETECTION_THRESHOLD:0}"
maximum-pool-size: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:32}"
maximum-pool-size: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:64}"
register-mbeans: "${SPRING_DATASOURCE_HIKARI_REGISTER_MBEANS:false}"
mybatis-plus:
@@ -66,7 +66,7 @@ queue:
in-memory:
queue-capacity: "${QUEUE_IN_MEMORY_QUEUE_CAPACITY:100000}"
stats:
print-interval-ms: "${QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:60000}"
print-interval-ms: "${QUEUE_IN_MEMORY_STATS_PRINT_INTERVAL_MS:10000}"
kafka:
bootstrap-servers: "${KAFKA_SERVERS:kafka:9092}"
ssl:
@@ -78,15 +78,15 @@ queue:
key-password: "${KAFKA_SSL_KEY_PASSWORD:}"
acks: "${KAFKA_ACKS:1}"
retries: "${KAFKA_RETRIES:1}"
compression-type: "${KAFKA_COMPRESSION_TYPE:lz4}" # none, gzip, snappy, lz4, zstd
batch-size: "${KAFKA_BATCH_SIZE:1048576}"
compression-type: "${KAFKA_COMPRESSION_TYPE:none}" # none, gzip, snappy, lz4, zstd
batch-size: "${KAFKA_BATCH_SIZE:16384}"
linger-ms: "${KAFKA_LINGER_MS:1}"
max-request-size: "${KAFKA_MAX_REQUEST_SIZE:1048576}"
max-in-flight-requests-per-connection: "${KAFKA_MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION:5}"
buffer-memory: "${BUFFER_MEMORY:33554432}"
replication-factor: "${QUEUE_KAFKA_REPLICATION_FACTOR:1}"
max-poll-interval-ms: "${QUEUE_KAFKA_MAX_POLL_INTERVAL_MS:300000}"
max-poll-records: "${QUEUE_KAFKA_MAX_POLL_RECORDS:10240}"
max-poll-records: "${QUEUE_KAFKA_MAX_POLL_RECORDS:8192}"
max-partition-fetch-bytes: "${QUEUE_KAFKA_MAX_PARTITION_FETCH_BYTES:16777216}"
fetch-max-bytes: "${QUEUE_KAFKA_FETCH_MAX_BYTES:134217728}"
request-timeout-ms: "${QUEUE_KAFKA_REQUEST_TIMEOUT_MS:30000}"
@@ -128,7 +128,7 @@ redis:
standalone:
host: "${REDIS_HOST:redis}"
port: "${REDIS_PORT:6379}"
useDefaultClientConfig: "${REDIS_USE_DEFAULT_CLIENT_CONFIG:false}"
useDefaultClientConfig: "${REDIS_USE_DEFAULT_CLIENT_CONFIG:true}"
clientName: "${REDIS_CLIENT_NAME:standalone}"
commandTimeout: "${REDIS_CLIENT_COMMAND_TIMEOUT:30000}"
shutdownTimeout: "${REDIS_CLIENT_SHUTDOWN_TIMEOUT:1000}"
@@ -146,9 +146,9 @@ redis:
db: "${REDIS_DB:0}"
password: "${REDIS_PASSWORD:sanbing}"
pool_config:
maxTotal: "${REDIS_POOL_CONFIG_MAX_TOTAL:128}"
maxIdle: "${REDIS_POOL_CONFIG_MAX_IDLE:64}"
minIdle: "${REDIS_POOL_CONFIG_MIN_IDLE:16}"
maxTotal: "${REDIS_POOL_CONFIG_MAX_TOTAL:256}"
maxIdle: "${REDIS_POOL_CONFIG_MAX_IDLE:128}"
minIdle: "${REDIS_POOL_CONFIG_MIN_IDLE:64}"
testOnBorrow: "${REDIS_POOL_CONFIG_TEST_ON_BORROW:false}"
testOnReturn: "${REDIS_POOL_CONFIG_TEST_ON_RETURN:false}"
testWhileIdle: "${REDIS_POOL_CONFIG_TEST_WHILE_IDLE:true}"
@@ -164,10 +164,23 @@ service:
type: "${SERVICE_TYPE:monolith}"
# 可自定义的服务ID如果不指定则默认为HOSTNAME
id: "${SERVICE_ID:}"
protocols:
protocol:
sessions:
default-inactivity-timeout-in-sec: "${PROTOCOLS_SESSIONS_DEFAULT_INACTIVITY_TIMEOUT_IN_SEC:600}"
default-state-check-interval-in-sec: "${PROTOCOLS_SESSIONS_DEFAULT_STATE_CHECK_INTERVAL_IN_SEC:60}"
rpc:
enabled: "${SERVICE_PROTOCOL_RPC_ENABLED:true}"
port: "${SERVICE_PROTOCOL_RPC_PORT:9090}"
boss: "${SERVICE_PROTOCOL_RPC_BOSS:4}"
worker: "${SERVICE_PROTOCOL_RPC_WORKER:64}"
so-rcvbuf: "${SERVICE_PROTOCOL_RPC_SO_RCVBUF:65535}"
so-sndbuf: "${SERVICE_PROTOCOL_RPC_SO_SNDBUF:65535}"
no-delay: "${SERVICE_PROTOCOL_RPC_NO_DELAY:true}"
user-thread-pool-size: "${SERVICE_PROTOCOL_RPC_USER_THREAD_POOL_SIZE:1024}"
max-inbound-message-size: "${SERVICE_PROTOCOL_RPC_MAX_INBOUND_MESSAGE_SIZE:33554432}"
max-concurrent-calls-per-connection: "${SERVICE_PROTOCOL_MAX_CONCURRENT_CALLS_PER_CONNECTION:4}"
client-max-keep-alive-time-sec: "${SERVICE_PROTOCOL_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:30}"
protocols:
yunkuaichongV150:
enabled: "${PROTOCOLS_YUNKUAICHONGV150_ENABLED:true}"
listener:
@@ -199,7 +212,7 @@ service:
# 以下配置只有在service.type为protocol时且jcpp-partition为false时才生效
bootstrap-servers: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_SERVERS:kafka:9092}"
acks: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_ACKS:1}"
# # 可选 protobuf推荐、json
# 可选 protobuf推荐、json
encoder: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_ENCODER:protobuf}"
retries: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_RETRIES:1}"
compression-type: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_COMPRESSION_TYPE:lz4}" # none, gzip, snappy, lz4, zstd
@@ -207,9 +220,52 @@ service:
linger-ms: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_LINGER_MS:0}"
buffer-memory: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_BUFFER_MEMORY:33554432}"
other-properties: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_QUEUE_KAFKA_OTHER_PROPERTIES:}"
yunkuaichongV160:
enabled: "${PROTOCOLS_YUNKUAICHONGV150_ENABLED:true}"
listener:
tcp:
bind-address: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BIND_ADDRESS:0.0.0.0}"
bind-port: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BIND_PORT:38002}"
boss-group-thread_count: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BOSS_GROUP_THREADS:4}"
worker-group-thread-count: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_WORKER_GROUP_THREADS:16}"
so-keep-alive: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_KEEPALIVE:true}"
so-backlog: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_BACKLOG:128}"
so-rcvbuf: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_RCVBUF:65536}"
so-sndbuf: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_SNDBUF:65536}"
nodelay: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_NODELAY:true}"
handler:
idle-timeout-seconds: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_HANDLER_IDLE_TIMEOUT_SECONDS:600}"
max_connections: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_HANDLER_MAX_CONNECTIONS:100000}"
# 默认为二进制类型的拆包器
# 可选JSON类型的拆包器 "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:JSON}"
# 可选纯文本类型的拆包器 "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:TEXT;maxFrameLength:128;stripDelimiter:true;messageSeparator:null;charsetName:UTF-8}"
configuration: "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:BINARY;decoder:sanbing.jcpp.protocol.listener.tcp.decoder.JCPPLengthFieldBasedFrameDecoder;byteOrder:LITTLE_ENDIAN;head:68;lengthFieldOffset:1;lengthFieldLength:1;lengthAdjustment:2;initialBytesToStrip:0}"
forwarder:
# 如果是单体服务可选kafka、memory未来计划扩展RocketMQ, GRpc、REST
type: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_TYPE:memory}"
memory:
topic: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_MEMORY_TOPIC:protocol_uplink}"
kafka:
topic: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_TOPIC:protocol_uplink}"
jcpp-partition: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_JCPP_PARTITION:true}" # 是否利用JCPP的分片框架
# 以下配置只有在service.type为protocol时且jcpp-partition为false时才生效
bootstrap-servers: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_SERVERS:kafka:9092}"
acks: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_ACKS:1}"
# 可选 protobuf推荐、json
encoder: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_ENCODER:protobuf}"
retries: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_RETRIES:1}"
compression-type: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_COMPRESSION_TYPE:lz4}" # none, gzip, snappy, lz4, zstd
batch-size: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_BATCH_SIZE:16384}"
linger-ms: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_LINGER_MS:0}"
buffer-memory: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_BUFFER_MEMORY:33554432}"
other-properties: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_QUEUE_KAFKA_OTHER_PROPERTIES:}"
thread-pool:
sharding:
hash_function_name: "${THREAD_POOL_SHARDING_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:8}"
stats-print-interval-ms: "${THREAD_POOL_SHARDING_STATS_PRINT_INTERVAL_MS:10000}"
downlink:
rpc:
type: "${DOWNLINK_RPC_TYPE:grpc}" # rest or grpc

View File

@@ -46,6 +46,12 @@
<AppenderRef ref="ROLLING_FILE"/>
</AsyncLogger>
<AsyncLogger name="sanbing.jcpp.app" level="${env:APP_LOG_LEVEL:-TRACE}"
additivity="false" includeLocation="false">
<AppenderRef ref="CONSOLE"/>
<AppenderRef ref="ROLLING_FILE"/>
</AsyncLogger>
<AsyncLogger name="sanbing.jcpp.protocol" level="${env:PROTOCOLS_LOG_LEVEL:-TRACE}"
additivity="false" includeLocation="false">
<AppenderRef ref="CONSOLE"/>

View File

@@ -1,2 +1,4 @@
spring.datasource.url=jdbc:postgresql://127.0.0.1:5432/jcpp
service.protocols.yunkuaichongV150.listener.tcp.bind-port=0
spring.datasource.url=jdbc:postgresql://testenv:30135/jcpp
service.protocols.yunkuaichongV150.listener.tcp.bind-port=0
service.protocols.yunkuaichongV160.listener.tcp.bind-port=0
service.protocol.rpc.port=0

View File

@@ -0,0 +1,31 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.app.adapter;
import jakarta.annotation.Resource;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import sanbing.jcpp.app.service.PileProtocolService;
import java.math.BigDecimal;
/**
* @author baigod
*/
@RestController
public class TestController {
@Resource
private PileProtocolService pileProtocolService;
@GetMapping("/api/startCharge")
public ResponseEntity<String> startCharge() {
pileProtocolService.startCharge("20231212000010", "01", new BigDecimal("50"), "12345678901234567890");
return ResponseEntity.ok("success");
}
}

View File

@@ -29,7 +29,11 @@ public class PileSession implements Serializable {
private String nodeId;
private String nodeWebapiIpPort;
private String nodeIp;
private int nodeRestPort;
private int nodeGrpcPort;
public PileSession(UUID pileId, String pileCode, String protocolName) {
this.pileId = pileId;
@@ -45,13 +49,17 @@ public class PileSession implements Serializable {
@JsonProperty("protocolSessionId") UUID protocolSessionId,
@JsonProperty("remoteAddress") String remoteAddress,
@JsonProperty("nodeId") String nodeId,
@JsonProperty("nodeWebapiIpPort") String nodeWebapiIpPort) {
@JsonProperty("nodeIp") String nodeIp,
@JsonProperty("nodeRestPort") int nodeRestPort,
@JsonProperty("nodeGrpcPort") int nodeGrpcPort) {
this.pileId = pileId;
this.pileCode = pileCode;
this.protocolName = protocolName;
this.protocolSessionId = protocolSessionId;
this.remoteAddress = remoteAddress;
this.nodeId = nodeId;
this.nodeWebapiIpPort = nodeWebapiIpPort;
this.nodeIp = nodeIp;
this.nodeRestPort = nodeRestPort;
this.nodeGrpcPort = nodeGrpcPort;
}
}

View File

@@ -4,12 +4,71 @@
*/
package sanbing.jcpp.app.service;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import sanbing.jcpp.app.data.PileSession;
import sanbing.jcpp.app.service.cache.session.PileSessionCacheKey;
import sanbing.jcpp.infrastructure.cache.CacheValueWrapper;
import sanbing.jcpp.infrastructure.cache.TransactionalCache;
import sanbing.jcpp.infrastructure.queue.discovery.ServiceInfoProvider;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.adapter.DownlinkController;
import java.util.UUID;
/**
* @author baigod
*/
public interface DownlinkCallService {
@Slf4j
public abstract class DownlinkCallService {
void sendDownlinkMessage(DownlinkRestMessage.Builder downlinkMessageBuilder, String pileCode);
@Resource
protected ServiceInfoProvider serviceInfoProvider;
@Resource
protected DownlinkController downlinkController;
@Resource
protected TransactionalCache<PileSessionCacheKey, PileSession> pileSessionCache;
@Value("${cache.type}")
protected String cacheType;
public void sendDownlinkMessage(DownlinkRequestMessage.Builder downlinkMessageBuilder, String pileCode) {
CacheValueWrapper<PileSession> pileSessionCacheValueWrapper = pileSessionCache.get(new PileSessionCacheKey(pileCode));
if (pileSessionCacheValueWrapper == null) {
log.warn("充电桩会话不存在 {}", pileCode);
return;
}
PileSession pileSession = pileSessionCacheValueWrapper.get();
UUID protocolSessionId = pileSession.getProtocolSessionId();
if (downlinkMessageBuilder.getSessionIdMSB() == 0) {
downlinkMessageBuilder.setSessionIdMSB(protocolSessionId.getMostSignificantBits());
}
if (downlinkMessageBuilder.getSessionIdLSB() == 0) {
downlinkMessageBuilder.setSessionIdLSB(protocolSessionId.getLeastSignificantBits());
}
if(downlinkMessageBuilder.getProtocolName() == null){
downlinkMessageBuilder.setProtocolName(pileSession.getProtocolName());
}
if (serviceInfoProvider.isMonolith() &&
("caffeine".equalsIgnoreCase(cacheType)) || serviceInfoProvider.getServiceId().equalsIgnoreCase(pileSession.getNodeId())) {
downlinkController.onDownlink(downlinkMessageBuilder.build())
.setResultHandler(result -> log.debug("下行消息发送完成"));
} else {
_sendDownlinkMessage(downlinkMessageBuilder.build(), pileSession);
}
}
protected abstract void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession);
}

View File

@@ -7,6 +7,8 @@ package sanbing.jcpp.app.service;
import sanbing.jcpp.infrastructure.queue.Callback;
import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
import java.math.BigDecimal;
/**
* @author baigod
*/
@@ -63,4 +65,9 @@ public interface PileProtocolService {
* 交易记录上报
*/
void onTransactionRecord(UplinkQueueMessage uplinkQueueMessage, Callback callback);
/**
* 启动充电
*/
void startCharge(String pileCode, String gunCode, BigDecimal limitYuan, String orderNo);
}

View File

@@ -0,0 +1,296 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.app.service.grpc;
import com.google.common.net.HostAndPort;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
import io.grpc.ManagedChannel;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import io.grpc.netty.shaded.io.netty.channel.ChannelOption;
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioSocketChannel;
import io.grpc.stub.StreamObserver;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import sanbing.jcpp.infrastructure.queue.discovery.ServiceInfoProvider;
import sanbing.jcpp.infrastructure.util.async.JCPPThreadFactory;
import sanbing.jcpp.infrastructure.util.mdc.MDCUtils;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
import sanbing.jcpp.proto.gen.ProtocolInterfaceGrpc;
import sanbing.jcpp.proto.gen.ProtocolInterfaceGrpc.ProtocolInterfaceStub;
import sanbing.jcpp.proto.gen.ProtocolProto.*;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import static sanbing.jcpp.infrastructure.proto.ProtoConverter.toTracerProto;
/**
* @author baigod
*/
@Component
@Slf4j
public class DownlinkGrpcClient {
@Value("${downlink.rpc.grpc.netty.event_loop:}")
private Integer rpcNettyEventLoop;
@Value("${downlink.rpc.grpc.netty.so_sndbuf:65535}")
private Integer rpcNettySoSndbuf;
@Value("${downlink.rpc.grpc.netty.so_rcvbuf:65535}")
private Integer rpcNettySoRcvbuf;
@Value("${downlink.rpc.grpc.netty.no_delay:true}")
private boolean rpcNoDelay;
@Value("${downlink.rpc.grpc.netty.max_inbound_message_size:33554432}")
private Integer rpcMaxInboundMessageSize;
@Value("${downlink.rpc.grpc.keep_alive_time_sec:300}")
private int keepAliveTimeSec;
@Value("${downlink.rpc.grpc.max_records_size:102400}")
private int maxRecordsSize;
@Value("${downlink.rpc.grpc.batch_records_count:1024}")
private int batchRecordsCount;
@Value("${downlink.rpc.grpc.no_read_records_sleep:25}")
private long noRecordsSleepInterval;
@Value("${downlink.rpc.grpc.records_ttl:600000}")
private long recordsTtl;
@Value("${downlink.rpc.grpc.max_reconnect_times:10}")
private int maxReconnectTimes;
@Resource
ServiceInfoProvider serviceInfoProvider;
private final Map<HostAndPort, ManagedChannel> channelMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, StreamObserver<RequestMsg>> inputStreamMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, LinkedBlockingQueue<RequestMsg>> queueMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, ReentrantLock> msgHandleLocksMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, ExecutorService> msgHandleExecutorMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, AtomicInteger> connectErrTimesMap = new ConcurrentHashMap<>();
private final Map<HostAndPort, Boolean> initializedMap = new ConcurrentHashMap<>();
private ScheduledExecutorService downlinkMsgsExecutor;
@PostConstruct
public void init() {
downlinkMsgsExecutor = Executors.newSingleThreadScheduledExecutor(JCPPThreadFactory.forName("downlink-msgs-executor"));
// 每秒进行一次连接检查与线程初始化
downlinkMsgsExecutor.scheduleWithFixedDelay(() -> {
queueMap.forEach((key, queue) -> {
ManagedChannel managedChannel = channelMap.get(key);
if (managedChannel == null) {
connect(key);
}
msgHandleExecutorMap.computeIfAbsent(key, hostAndPort ->
Executors.newFixedThreadPool(1, JCPPThreadFactory.forName("downlink-handle-threads-" + hostAndPort)))
.execute(new TracerRunnable(() -> {
while (Boolean.TRUE.equals(initializedMap.computeIfAbsent(key, k -> Boolean.FALSE))) {
try {
handleMsgs(key, queue);
} catch (Exception e) {
log.error("Failed to process messages handling!", e);
}
}
}));
});
}, 0, 1, TimeUnit.SECONDS);
}
private void handleMsgs(HostAndPort key, LinkedBlockingQueue<RequestMsg> queue) {
StreamObserver<RequestMsg> inputStream = inputStreamMap.get(key);
if (inputStream == null) {
return;
}
long acceptTs = System.currentTimeMillis() - recordsTtl;
List<RequestMsg> downlinkMsgs = new ArrayList<>(batchRecordsCount);
queue.drainTo(downlinkMsgs, batchRecordsCount);
for (RequestMsg msg : downlinkMsgs) {
long ts = msg.getTs();
if (ts > 0 && ts < acceptTs) {
log.warn("[{}] 消息过期,直接丢弃 {}", key, ts);
continue;
}
ReentrantLock lock = msgHandleLocksMap.computeIfAbsent(key, hostAndPort -> new ReentrantLock());
lock.lock();
try {
inputStream.onNext(msg);
} finally {
lock.unlock();
}
}
if (downlinkMsgs.isEmpty()) {
try {
Thread.sleep(noRecordsSleepInterval);
} catch (InterruptedException e) {
log.warn("Sleep interrupted!", e);
}
} else {
downlinkMsgs.clear();
}
}
@PreDestroy
public void destroy() {
log.info("Starting Grpc destroying process");
initializedMap.replaceAll((hostAndPort, aBoolean) -> Boolean.FALSE);
try {
downlinkMsgsExecutor.shutdownNow();
msgHandleExecutorMap.values().forEach(ExecutorService::shutdownNow);
inputStreamMap.values().forEach(StreamObserver::onCompleted);
channelMap.values().forEach(ManagedChannel::shutdownNow);
} catch (Exception e) {
log.error("Exception during disconnect", e);
}
}
public void connect(HostAndPort hostAndPort) {
log.info("[{}] Create new Grpc Client Channel!", hostAndPort);
ManagedChannel managedChannel = NettyChannelBuilder.forAddress(hostAndPort.getHost(), hostAndPort.getPort())
.eventLoopGroup(new NioEventLoopGroup(Optional.ofNullable(rpcNettyEventLoop).orElse(Runtime.getRuntime().availableProcessors() * 2)))
.compressorRegistry(CompressorRegistry.getDefaultInstance())
.decompressorRegistry(DecompressorRegistry.getDefaultInstance())
.withOption(ChannelOption.SO_SNDBUF, rpcNettySoSndbuf)
.withOption(ChannelOption.SO_RCVBUF, rpcNettySoRcvbuf)
.withOption(ChannelOption.TCP_NODELAY, rpcNoDelay)
.maxInboundMessageSize(rpcMaxInboundMessageSize)
.channelType(NioSocketChannel.class)
.directExecutor()
.keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS)
.usePlaintext()
.keepAliveTime(5, TimeUnit.MINUTES) // Change to a larger value, e.g. 5min.
.keepAliveTimeout(10, TimeUnit.SECONDS) // Change to a larger value, e.g. 10s.
.keepAliveWithoutCalls(true)// You should normally avoid enabling this.
.defaultLoadBalancingPolicy("round_robin")
.build();
ManagedChannel remove = channelMap.remove(hostAndPort);
if (remove != null) {
channelMap.get(hostAndPort).shutdownNow();
}
channelMap.put(hostAndPort, managedChannel);
ProtocolInterfaceStub stub = ProtocolInterfaceGrpc.newStub(managedChannel);
StreamObserver<RequestMsg> streamObserver = stub.onDownlink(new StreamObserver<>() {
@Override
public void onNext(ResponseMsg responseMsg) {
TracerProto tracerProto = responseMsg.getTracer();
TracerContextUtil.newTracer(tracerProto.getId(), tracerProto.getOrigin(), tracerProto.getTs());
MDCUtils.recordTracer();
if (responseMsg.hasConnectResponseMsg()) {
log.info("[{}] Grpc 接收到通信层连接反馈 {}", hostAndPort, responseMsg.getConnectResponseMsg());
if (ConnectResponseCode.ACCEPTED.equals(responseMsg.getConnectResponseMsg().getResponseCode())) {
initializedMap.put(hostAndPort, Boolean.TRUE);
} else {
onError(new RuntimeException(responseMsg.getConnectResponseMsg().getErrorMsg()));
}
}
if (responseMsg.hasDownlinkResponseMsg()) {
DownlinkResponseMessage downlinkResponseMsg = responseMsg.getDownlinkResponseMsg();
if (!downlinkResponseMsg.getSuccess()) {
log.info("[{}] Grpc 下行数据发生错误回复 {}", hostAndPort, downlinkResponseMsg);
}
}
}
@Override
public void onError(Throwable t) {
log.warn("[{}] Grpc 客户端异常 {}", hostAndPort, t.getMessage());
ExecutorService executorService = msgHandleExecutorMap.get(hostAndPort);
if (executorService != null) {
executorService.shutdownNow();
msgHandleExecutorMap.remove(hostAndPort);
}
ManagedChannel remove = channelMap.remove(hostAndPort);
if (remove != null) {
remove.shutdownNow();
}
if (connectErrTimesMap.computeIfAbsent(hostAndPort, k -> new AtomicInteger()).incrementAndGet() >= maxReconnectTimes) {
LinkedBlockingQueue<RequestMsg> queue = queueMap.remove(hostAndPort);
if (queue != null) {
queue.clear();
}
connectErrTimesMap.remove(hostAndPort);
log.info("[{}] Grpc 客户端重连异常超过{}次,不再重连", hostAndPort, maxReconnectTimes);
}
}
@Override
public void onCompleted() {
log.info("[{}] The Grpc connection was closed!", hostAndPort);
}
});
streamObserver.onNext(RequestMsg.newBuilder()
.setTracer(toTracerProto())
.setConnectRequestMsg(ConnectRequestMsg.newBuilder()
.setNodeId(serviceInfoProvider.getServiceId())
.build())
.build());
inputStreamMap.put(hostAndPort, streamObserver);
}
/**
* 发送下行请求
*/
public void sendDownlinkRequest(HostAndPort hostAndPort, RequestMsg requestMsg) {
queueMap.computeIfAbsent(hostAndPort, k -> new LinkedBlockingQueue<>(maxRecordsSize)).add(requestMsg);
}
}

View File

@@ -1,93 +0,0 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.app.service.impl;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import sanbing.jcpp.app.data.PileSession;
import sanbing.jcpp.app.service.DownlinkCallService;
import sanbing.jcpp.app.service.cache.session.PileSessionCacheKey;
import sanbing.jcpp.infrastructure.cache.CacheValueWrapper;
import sanbing.jcpp.infrastructure.cache.TransactionalCache;
import sanbing.jcpp.infrastructure.queue.discovery.ServiceInfoProvider;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.protocol.adapter.DownlinkController;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.*;
/**
* @author baigod
*/
@Service
@Slf4j
public class DefaultDownlinkCallService implements DownlinkCallService {
@Resource
RestTemplate downlinkRestTemplate;
@Resource
ServiceInfoProvider serviceInfoProvider;
@Resource
DownlinkController downlinkController;
@Resource
TransactionalCache<PileSessionCacheKey, PileSession> pileSessionCache;
@Override
public void sendDownlinkMessage(DownlinkRestMessage.Builder downlinkMessageBuilder, String pileCode) {
if (serviceInfoProvider.isMonolith()) {
downlinkController.onDownlink(downlinkMessageBuilder.build())
.setResultHandler(result -> log.debug("下行消息发送完成"));
} else {
try {
CacheValueWrapper<PileSession> pileSessionCacheValueWrapper = pileSessionCache.get(new PileSessionCacheKey(pileCode));
if (pileSessionCacheValueWrapper == null) {
log.warn("充电桩会话不存在 {}", pileCode);
return;
}
PileSession pileSession = pileSessionCacheValueWrapper.get();
invokeDownlinkRestApi(downlinkMessageBuilder.build(), pileSession.getNodeWebapiIpPort());
} catch (RestClientException e) {
log.error("下行消息发送异常", e);
}
}
}
private void invokeDownlinkRestApi(DownlinkRestMessage downlinkRestMessage, String nodeWebapiIpPort) {
HttpHeaders headers = new HttpHeaders();
headers.add(JCPP_TRACER_ID, TracerContextUtil.getCurrentTracer().getTraceId());
headers.add(JCPP_TRACER_ORIGIN, TracerContextUtil.getCurrentTracer().getOrigin());
headers.add(JCPP_TRACER_TS, String.valueOf(TracerContextUtil.getCurrentTracer().getTracerTs()));
headers.setContentType(MediaType.parseMediaType("application/x-protobuf"));
HttpEntity<DownlinkRestMessage> entity = new HttpEntity<>(downlinkRestMessage, headers);
try {
ResponseEntity<?> response = downlinkRestTemplate.postForEntity("http://" + nodeWebapiIpPort + "/api/onDownlink",
entity, ResponseEntity.class);
log.info("下行消息发送成功 {}", response);
} catch (RestClientException e) {
log.error("下行消息发送失败 {}", downlinkRestMessage, e);
throw new RuntimeException(e);
}
}
}

View File

@@ -59,7 +59,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
log.debug("查询到充电桩信息 {}", pile);
// 构造下行回复
DownlinkRestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, loginRequest.getPileCode());
DownlinkRequestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, loginRequest.getPileCode());
downlinkMessageBuilder.setDownlinkCmd(DownlinkCmdEnum.LOGIN_ACK.name());
if (pile != null) {
@@ -67,7 +67,9 @@ public class DefaultPileProtocolService implements PileProtocolService {
cacheSession(uplinkQueueMessage, pile,
loginRequest.getRemoteAddress(),
loginRequest.getNodeId(),
loginRequest.getNodeWebapiIpPort());
loginRequest.getNodeHostAddress(),
loginRequest.getNodeRestPort(),
loginRequest.getNodeGrpcPort());
downlinkMessageBuilder.setLoginResponse(LoginResponse.newBuilder()
.setSuccess(true)
@@ -98,16 +100,26 @@ public class DefaultPileProtocolService implements PileProtocolService {
cacheSession(uplinkQueueMessage, pile,
heartBeatRequest.getRemoteAddress(),
heartBeatRequest.getNodeId(),
heartBeatRequest.getNodeWebapiIpPort());
heartBeatRequest.getNodeHostAddress(),
heartBeatRequest.getNodeRestPort(),
heartBeatRequest.getNodeGrpcPort());
}
}
private void cacheSession(UplinkQueueMessage uplinkQueueMessage, Pile pile, String remoteAddress, String nodeId, String nodeWebapiIpPort) {
private void cacheSession(UplinkQueueMessage uplinkQueueMessage,
Pile pile,
String remoteAddress,
String nodeId,
String nodeIp,
int restPort,
int grpcPort) {
PileSession pileSession = new PileSession(pile.getId(), pile.getPileCode(), uplinkQueueMessage.getProtocolName());
pileSession.setProtocolSessionId(new UUID(uplinkQueueMessage.getSessionIdMSB(), uplinkQueueMessage.getSessionIdLSB()));
pileSession.setRemoteAddress(remoteAddress);
pileSession.setNodeId(nodeId);
pileSession.setNodeWebapiIpPort(nodeWebapiIpPort);
pileSession.setNodeIp(nodeIp);
pileSession.setNodeRestPort(restPort);
pileSession.setNodeGrpcPort(grpcPort);
pileSessionCache.put(new PileSessionCacheKey(pile.getPileCode()), pileSession);
}
@@ -122,7 +134,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
// todo 默认校验成功,后续查库校验
assert pricingId > 0;
DownlinkRestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
DownlinkRequestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
downlinkMessageBuilder.setDownlinkCmd(DownlinkCmdEnum.VERIFY_PRICING_ACK.name());
downlinkMessageBuilder.setVerifyPricingResponse(VerifyPricingResponse.newBuilder()
.setSuccess(true)
@@ -167,7 +179,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
model.setPeriodsList(periods);
// 构造下行计费
DownlinkRestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
DownlinkRequestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
downlinkMessageBuilder.setDownlinkCmd(DownlinkCmdEnum.QUERY_PRICING_ACK.name());
downlinkMessageBuilder.setQueryPricingResponse(QueryPricingResponse.newBuilder()
.setPileCode(pileCode)
@@ -236,7 +248,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
String pileCode = transactionRecord.getPileCode();
// 构造下行计费
DownlinkRestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
DownlinkRequestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, pileCode);
downlinkMessageBuilder.setDownlinkCmd(DownlinkCmdEnum.TRANSACTION_RECORD.name());
downlinkMessageBuilder.setTransactionRecordAck(TransactionRecordAck.newBuilder()
.setTradeNo(tradeNo)
@@ -248,6 +260,31 @@ public class DefaultPileProtocolService implements PileProtocolService {
callback.onSuccess();
}
@Override
public void startCharge(String pileCode, String gunCode, BigDecimal limitYuan, String orderNo) {
UUID messageId = UUID.randomUUID();
UUID requestId = UUID.randomUUID();
DownlinkRequestMessage.Builder downlinkRequestMessageBuilder = DownlinkRequestMessage.newBuilder()
.setMessageIdMSB(messageId.getMostSignificantBits())
.setMessageIdLSB(messageId.getLeastSignificantBits())
.setPileCode(pileCode)
.setRequestIdMSB(requestId.getMostSignificantBits())
.setRequestIdLSB(requestId.getLeastSignificantBits())
.setDownlinkCmd(DownlinkCmdEnum.REMOTE_START_CHARGING.name())
.setRemoteStartChargingRequest(RemoteStartChargingRequest.newBuilder()
.setPileCode(pileCode)
.setGunCode(gunCode)
.setLimitYuan(limitYuan.toPlainString())
.setTradeNo(orderNo)
.build());
downlinkCallService.sendDownlinkMessage(downlinkRequestMessageBuilder, pileCode);
}
private static Period createPeriod(int sn, LocalTime beginTime, LocalTime endTime, PricingModelFlag flag) {
Period period = new Period();
period.setSn(sn);
@@ -257,9 +294,9 @@ public class DefaultPileProtocolService implements PileProtocolService {
return period;
}
private DownlinkRestMessage.Builder createDownlinkMessageBuilder(UplinkQueueMessage uplinkQueueMessage, String pileCode) {
private DownlinkRequestMessage.Builder createDownlinkMessageBuilder(UplinkQueueMessage uplinkQueueMessage, String pileCode) {
UUID messageId = UUID.randomUUID();
DownlinkRestMessage.Builder builder = DownlinkRestMessage.newBuilder();
DownlinkRequestMessage.Builder builder = DownlinkRequestMessage.newBuilder();
builder.setMessageIdMSB(messageId.getLeastSignificantBits());
builder.setMessageIdLSB(messageId.getLeastSignificantBits());
builder.setPileCode(pileCode);

View File

@@ -0,0 +1,48 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.app.service.impl;
import com.google.common.net.HostAndPort;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import sanbing.jcpp.app.data.PileSession;
import sanbing.jcpp.app.service.DownlinkCallService;
import sanbing.jcpp.app.service.grpc.DownlinkGrpcClient;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.RequestMsg;
import static sanbing.jcpp.infrastructure.proto.ProtoConverter.toTracerProto;
/**
* @author baigod
*/
@Service
@Slf4j
@ConditionalOnExpression("'${downlink.rpc.type:null}'=='grpc'")
public class GrpcDownlinkCallService extends DownlinkCallService {
@Resource
DownlinkGrpcClient downlinkGrpcClient;
@Override
protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession) {
try {
RequestMsg requestMsg = RequestMsg.newBuilder()
.setTs(System.currentTimeMillis())
.setTracer(toTracerProto())
.setDownlinkRequestMessage(downlinkMessage)
.build();
downlinkGrpcClient.sendDownlinkRequest(HostAndPort.fromParts(pileSession.getNodeIp(), pileSession.getNodeGrpcPort()),
requestMsg);
} catch (Exception e) {
log.error("下行消息发送异常", e);
}
}
}

View File

@@ -0,0 +1,64 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.app.service.impl;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import sanbing.jcpp.app.data.PileSession;
import sanbing.jcpp.app.service.DownlinkCallService;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.*;
/**
* @author baigod
*/
@Service
@Slf4j
@ConditionalOnExpression("'${downlink.rpc.type:null}'=='rest'")
public class RestDownlinkCallService extends DownlinkCallService {
@Resource
RestTemplate downlinkRestTemplate;
@Override
protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession) {
try {
invokeDownlinkRestApi(downlinkMessage, pileSession.getNodeIp(), pileSession.getNodeRestPort());
} catch (RestClientException e) {
log.error("下行消息发送异常", e);
}
}
private void invokeDownlinkRestApi(DownlinkRequestMessage downlinkRequestMessage, String nodeWebapiIpPort, int nodeRestPort) {
HttpHeaders headers = new HttpHeaders();
headers.add(JCPP_TRACER_ID, TracerContextUtil.getCurrentTracer().getTraceId());
headers.add(JCPP_TRACER_ORIGIN, TracerContextUtil.getCurrentTracer().getOrigin());
headers.add(JCPP_TRACER_TS, String.valueOf(TracerContextUtil.getCurrentTracer().getTracerTs()));
headers.setContentType(MediaType.parseMediaType("application/x-protobuf"));
HttpEntity<DownlinkRequestMessage> entity = new HttpEntity<>(downlinkRequestMessage, headers);
try {
ResponseEntity<?> response = downlinkRestTemplate.postForEntity("http://" + nodeWebapiIpPort + ":" + nodeRestPort + "/api/onDownlink",
entity, ResponseEntity.class);
log.debug("下行消息发送成功 {}", response);
} catch (RestClientException e) {
log.error("下行消息发送失败 {}", downlinkRequestMessage, e);
throw new RuntimeException(e);
}
}
}

View File

@@ -35,16 +35,12 @@ import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
/**
@@ -134,82 +130,108 @@ public class ProtocolUplinkConsumerService extends AbstractConsumerService imple
PendingMsgHolder pendingMsgHolder = new PendingMsgHolder();
Future<?> packSubmitFuture = consumersExecutor.submit(new TracerRunnable(() ->
orderedMsgList.forEach(element -> {
UUID id = element.getUuid();
ProtoQueueMsg<UplinkQueueMessage> msg = element.getMsg();
tracer(msg);
log.trace("[{}] Creating main callback for message: {}", id, msg.getValue());
log.trace("[{}] Creating PackCallback for message: {}", id, msg.getValue());
Callback callback = new PackCallback<>(id, ctx);
try {
UplinkQueueMessage uplinkQueueMsg = msg.getValue();
pendingMsgHolder.setUplinkQueueMessage(uplinkQueueMsg);
if (uplinkQueueMsg.hasLoginRequest()) {
pileProtocolService.pileLogin(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasHeartBeatRequest()) {
pileProtocolService.heartBeat(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasVerifyPricingRequest()) {
pileProtocolService.verifyPricing(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasQueryPricingRequest()) {
pileProtocolService.queryPricing(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasGunRunStatusProto()) {
pileProtocolService.postGunRunStatus(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasChargingProgressProto()) {
pileProtocolService.postChargingProgress(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasSetPricingResponse()) {
pileProtocolService.onSetPricingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasRemoteStartChargingResponse()) {
pileProtocolService.onRemoteStartChargingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasRemoteStopChargingResponse()) {
pileProtocolService.onRemoteStopChargingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasTransactionRecord()) {
pileProtocolService.onTransactionRecord(uplinkQueueMsg, callback);
} else {
callback.onSuccess();
}
if (statsEnabled) {
stats.log(uplinkQueueMsg);
}
pendingMsgHolder.setUplinkQueueMessage(uplinkQueueMsg);
if (uplinkQueueMsg.hasLoginRequest()) {
pileProtocolService.pileLogin(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasHeartBeatRequest()) {
pileProtocolService.heartBeat(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasVerifyPricingRequest()) {
pileProtocolService.verifyPricing(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasQueryPricingRequest()) {
pileProtocolService.queryPricing(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasGunRunStatusProto()) {
pileProtocolService.postGunRunStatus(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasChargingProgressProto()) {
pileProtocolService.postChargingProgress(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasSetPricingResponse()) {
pileProtocolService.onSetPricingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasRemoteStartChargingResponse()) {
pileProtocolService.onRemoteStartChargingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasRemoteStopChargingResponse()) {
pileProtocolService.onRemoteStopChargingResponse(uplinkQueueMsg, callback);
} else if (uplinkQueueMsg.hasTransactionRecord()) {
pileProtocolService.onTransactionRecord(uplinkQueueMsg, callback);
} else {
callback.onSuccess();
}
} catch (Throwable e) {
log.warn("[{}] Failed to process message: {}", id, msg, e);
callback.onFailure(e);
}
}))
);
if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) {
if (!ctx.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) {
if (!packSubmitFuture.isDone()) {
packSubmitFuture.cancel(true);
UplinkQueueMessage lastSubmitMsg = pendingMsgHolder.getUplinkQueueMessage();
log.warn("Timeout to process message: {}", lastSubmitMsg);
}
if (log.isDebugEnabled()) {
ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue()));
ctx.getAckMap().forEach((id, msg) -> log.info("[{}] Timeout to process message: {}", id, msg.getValue()));
}
ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue()));
}
consumer.commit();
}
private void tracer(ProtoQueueMsg<UplinkQueueMessage> msg) {
if (Optional.ofNullable(msg.getHeaders().get(MSG_MD_PREFIX + JCPP_TRACER_ID))
.map(tracerId -> {
String origin = null;
byte[] tracerOrigin = msg.getHeaders().get(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN);
if (tracerOrigin != null) {
origin = ByteUtil.bytesToString(tracerOrigin);
}
long ts = System.currentTimeMillis();
byte[] tracerTs = msg.getHeaders().get(MSG_MD_PREFIX + MSG_MD_TS);
if (tracerTs != null) {
ts = ByteUtil.bytesToLong(tracerTs);
}
return TracerContextUtil.newTracer(ByteUtil.bytesToString(tracerId), origin, ts);
})
.isEmpty()) {
TracerContextUtil.newTracer();
}
TracerContextUtil.newTracer(ByteUtil.bytesToString(msg.getHeaders().get(MSG_MD_TRACER_ID)),
ByteUtil.bytesToString(msg.getHeaders().get(MSG_MD_TRACER_ORIGIN)),
ByteUtil.bytesToLong(msg.getHeaders().get(MSG_MD_TRACER_TS)));
MDCUtils.recordTracer();
}

View File

@@ -26,6 +26,14 @@
</properties>
<dependencies>
<dependency>
<groupId>sanbing</groupId>
<artifactId>jcpp-infrastructure-util</artifactId>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
@@ -34,6 +42,18 @@
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java-util</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
</dependency>
</dependencies>
<build>
<plugins>

View File

@@ -8,6 +8,8 @@ package sanbing.jcpp.infrastructure.proto;
import sanbing.jcpp.infrastructure.proto.model.PricingModel;
import sanbing.jcpp.infrastructure.proto.model.PricingModel.FlagPrice;
import sanbing.jcpp.infrastructure.proto.model.PricingModel.Period;
import sanbing.jcpp.infrastructure.util.trace.Tracer;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.*;
import java.util.Map;
@@ -17,6 +19,15 @@ import java.util.Map;
*/
public class ProtoConverter {
public static TracerProto toTracerProto() {
Tracer currentTracer = TracerContextUtil.getCurrentTracer();
return TracerProto.newBuilder()
.setId(currentTracer.getTraceId())
.setOrigin(currentTracer.getOrigin())
.setTs(currentTracer.getTracerTs())
.build();
}
public static PricingModelProto toPricingModel(PricingModel pricingModel) {
// 创建 PricingModelProto 实例
PricingModelProto.Builder builder = PricingModelProto.newBuilder();

View File

@@ -9,6 +9,43 @@ package infrastructureProto;
option java_package = "sanbing.jcpp.proto.gen";
option java_outer_classname = "ProtocolProto";
service ProtocolInterface {
rpc onDownlink(stream RequestMsg) returns (stream ResponseMsg) {}
}
message RequestMsg {
int64 ts = 1;
TracerProto tracer = 2;
ConnectRequestMsg connectRequestMsg = 10;
DownlinkRequestMessage downlinkRequestMessage = 11;
}
message ResponseMsg {
TracerProto tracer = 2;
ConnectResponseMsg connectResponseMsg = 12;
DownlinkResponseMessage downlinkResponseMsg = 13;
}
message ConnectResponseMsg {
ConnectResponseCode responseCode = 1;
string errorMsg = 2;
}
message ConnectRequestMsg {
string nodeId = 1;
}
enum ConnectResponseCode {
ACCEPTED = 0;
REFUSE = 1;
}
message TracerProto {
string id = 1;
string origin = 2;
int64 ts = 3;
}
message UplinkQueueMessage {
int64 messageIdMSB = 1;
int64 messageIdLSB = 2;
@@ -29,7 +66,7 @@ message UplinkQueueMessage {
TransactionRecord transactionRecord = 30;
}
message DownlinkRestMessage {
message DownlinkRequestMessage {
int64 messageIdMSB = 1;
int64 messageIdLSB = 2;
int64 sessionIdMSB = 3;
@@ -39,14 +76,19 @@ message DownlinkRestMessage {
optional int64 requestIdMSB = 8;
optional int64 requestIdLSB = 9;
optional bytes requestData = 10;
string downlinkCmd = 11;
LoginResponse loginResponse = 20;
VerifyPricingResponse verifyPricingResponse = 21;
QueryPricingResponse queryPricingResponse = 22;
SetPricingRequest setPricingRequest = 23;
RemoteStartChargingRequest remoteStartChargingRequest = 24;
RemoteStopChargingRequest remoteStopChargingRequest = 25;
TransactionRecordAck transactionRecordAck = 26;
string downlinkCmd = 20;
LoginResponse loginResponse = 21;
VerifyPricingResponse verifyPricingResponse = 22;
QueryPricingResponse queryPricingResponse = 23;
SetPricingRequest setPricingRequest = 24;
RemoteStartChargingRequest remoteStartChargingRequest = 25;
RemoteStopChargingRequest remoteStopChargingRequest = 26;
TransactionRecordAck transactionRecordAck = 27;
}
message DownlinkResponseMessage {
bool success = 1;
optional string error = 2;
}
message LoginRequest {
@@ -54,7 +96,9 @@ message LoginRequest {
string credential = 3;
string remoteAddress = 4;
string nodeId = 10;
string nodeWebapiIpPort = 11;
string nodeHostAddress = 11;
int32 nodeRestPort = 12;
int32 nodeGrpcPort = 13;
optional string additionalInfo = 20;
}
@@ -67,7 +111,9 @@ message HeartBeatRequest {
string pileCode = 3;
string remoteAddress = 4;
string nodeId = 10;
string nodeWebapiIpPort = 11;
string nodeHostAddress = 11;
int32 nodeRestPort = 12;
int32 nodeGrpcPort = 13;
optional string additionalInfo = 20;
}
@@ -186,7 +232,7 @@ message RemoteStartChargingRequest {
string pileCode = 4;
string gunCode = 5;
string tradeNo = 6;
int32 limitYuan = 7;
string limitYuan = 7;
optional string additionalInfo = 20;
}

View File

@@ -4,10 +4,11 @@
*/
package sanbing.jcpp.infrastructure.queue;
import com.google.protobuf.GeneratedMessage;
import lombok.Data;
@Data
public class ProtoQueueMsg<T extends com.google.protobuf.GeneratedMessageV3> implements QueueMsg {
public class ProtoQueueMsg<T extends GeneratedMessage> implements QueueMsg {
private final String key;
protected final T value;

View File

@@ -4,6 +4,8 @@
*/
package sanbing.jcpp.infrastructure.queue.common;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.*;
/**
* @author baigod
*/
@@ -11,5 +13,10 @@ public final class QueueConstants {
public static final String MSG_MD_PREFIX = "jcpp_";
public static final String MSG_MD_TS = "ts";
public static final String MSG_MD_TRACER_ID = MSG_MD_PREFIX + JCPP_TRACER_ID;
public static final String MSG_MD_TRACER_ORIGIN = MSG_MD_PREFIX + JCPP_TRACER_ORIGIN;
public static final String MSG_MD_TRACER_TS = MSG_MD_PREFIX + JCPP_TRACER_TS;
}

View File

@@ -19,7 +19,6 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
/**
@@ -42,15 +41,18 @@ public class DefaultServiceInfoProvider implements ServiceInfoProvider {
private ServiceInfo serviceInfo;
@Getter
private String serviceWebapiEndpoint;
private String hostAddress;
@Getter
@Value("${server.port}")
private String webapiPort;
private int restPort;
@Getter
@Value("${service.protocol.rpc.port:9090}")
private int grpcPort;
@PostConstruct
public void init() throws UnknownHostException {
if (!StringUtils.hasText(this.serviceId)) {
try {
this.serviceId = InetAddress.getLocalHost().getHostName();
@@ -58,10 +60,11 @@ public class DefaultServiceInfoProvider implements ServiceInfoProvider {
this.serviceId = RandomStringUtils.randomAlphabetic(10);
}
}
log.info("Current Service ID: {}", this.serviceId);
log.info("Current Service ID: {}", serviceId);
serviceWebapiEndpoint = InetAddress.getLocalHost().getHostAddress() + ":" + webapiPort;
log.info("Current Service HostAddress: {}", this.serviceWebapiEndpoint);
hostAddress = InetAddress.getLocalHost().getHostAddress();
log.info("Current Service HostAddress: {}, RestPort:{}, GrpcPort:{}", hostAddress, restPort, grpcPort);
if (serviceType.equalsIgnoreCase("monolith")) {
serviceTypes = List.of(ServiceType.values());
} else {
@@ -86,7 +89,7 @@ public class DefaultServiceInfoProvider implements ServiceInfoProvider {
public ServiceInfo generateNewServiceInfoWithCurrentSystemInfo() {
ServiceInfo.Builder builder = ServiceInfo.newBuilder()
.setServiceId(serviceId)
.addAllServiceTypes(serviceTypes.stream().map(ServiceType::name).collect(Collectors.toList()))
.addAllServiceTypes(serviceTypes.stream().map(ServiceType::name).toList())
.setSystemInfo(getCurrentSystemInfoProto());
return serviceInfo = builder.build();
}

View File

@@ -13,7 +13,11 @@ import sanbing.jcpp.proto.gen.ClusterProto;
public interface ServiceInfoProvider {
String getServiceId();
String getServiceWebapiEndpoint();
String getHostAddress();
int getRestPort();
int getGrpcPort();
String getServiceType();

View File

@@ -4,12 +4,13 @@
*/
package sanbing.jcpp.infrastructure.queue.processing;
import com.google.protobuf.GeneratedMessage;
import lombok.Getter;
import sanbing.jcpp.infrastructure.queue.ProtoQueueMsg;
import java.util.UUID;
public class IdMsgPair<T extends com.google.protobuf.GeneratedMessageV3> {
public class IdMsgPair<T extends GeneratedMessage> {
@Getter
final UUID uuid;
@Getter

View File

@@ -41,7 +41,6 @@ public class KafkaAppQueueFactory implements AppQueueFactory {
this.appAdmin = new KafkaAdmin(kafkaSettings, kafkaTopicConfigs.getAppConfigs());
}
@Override
public QueueConsumer<ProtoQueueMsg<UplinkQueueMessage>> createProtocolUplinkMsgConsumer() {
KafkaConsumerTemplate.KafkaConsumerTemplateBuilder<ProtoQueueMsg<UplinkQueueMessage>> consumerBuilder = KafkaConsumerTemplate.builder();

View File

@@ -34,7 +34,6 @@ public class MDCUtils {
}
return tracer.getTraceId();
}
public static void cleanTracer() {

View File

@@ -6,11 +6,15 @@ package sanbing.jcpp.infrastructure.util.trace;
import org.apache.commons.lang3.StringUtils;
import java.util.Optional;
/**
* Tracer上下文工具类
*/
public class TracerContextUtil {
public static final String DEFAULT_ORIGIN = "jcpp";
public static final String JCPP_TRACER_ID = "jcpp_tracer_id";
public static final String JCPP_TRACER_ORIGIN = "jcpp_tracer_origin";
public static final String JCPP_TRACER_TS = "jcpp_tracer_ts";
@@ -20,6 +24,8 @@ public class TracerContextUtil {
public static Tracer newTracer(String traceId, String origin) {
Tracer tracer;
origin = Optional.ofNullable(origin).orElse(DEFAULT_ORIGIN);
if (StringUtils.isEmpty(traceId)) {
tracer = new Tracer(TraceIdGenerator.generate(), origin);
} else {
@@ -34,13 +40,14 @@ public class TracerContextUtil {
public static Tracer newTracer(String traceId, String origin, long ts) {
final Tracer tracer;
origin = Optional.ofNullable(origin).orElse(DEFAULT_ORIGIN);
if (StringUtils.isEmpty(traceId)) {
tracer = new Tracer(TraceIdGenerator.generate(), origin, ts);
} else {
tracer = new Tracer(traceId, origin, ts);
}
TRACE_ID_CONTAINER.set(tracer);
return tracer;

View File

@@ -15,7 +15,7 @@ import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.domain.ProtocolSession;
import sanbing.jcpp.protocol.provider.ProtocolSessionRegistryProvider;
@@ -37,7 +37,7 @@ public class DownlinkController {
ProtocolSessionRegistryProvider protocolSessionRegistryProvider;
@PostMapping(value = "/onDownlink", consumes = "application/x-protobuf", produces = "application/x-protobuf")
public DeferredResult<ResponseEntity<String>> onDownlink(@RequestBody DownlinkRestMessage downlinkMsg) {
public DeferredResult<ResponseEntity<String>> onDownlink(@RequestBody DownlinkRequestMessage downlinkMsg) {
log.debug("收到REST下行请求 {}", downlinkMsg);
final DeferredResult<ResponseEntity<String>> response = new DeferredResult<>(onDownlinkTimeout,

View File

@@ -0,0 +1,182 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.adapter;
import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry;
import io.grpc.Server;
import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
import io.grpc.netty.shaded.io.netty.channel.ChannelOption;
import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
import io.grpc.stub.StreamObserver;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;
import sanbing.jcpp.infrastructure.util.mdc.MDCUtils;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.infrastructure.util.trace.TracerRunnable;
import sanbing.jcpp.proto.gen.ProtocolInterfaceGrpc.ProtocolInterfaceImplBase;
import sanbing.jcpp.proto.gen.ProtocolProto.*;
import sanbing.jcpp.protocol.domain.ProtocolSession;
import sanbing.jcpp.protocol.provider.ProtocolSessionRegistryProvider;
import javax.annotation.PreDestroy;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import static sanbing.jcpp.infrastructure.proto.ProtoConverter.toTracerProto;
import static sanbing.jcpp.infrastructure.util.config.ThreadPoolConfiguration.JCPP_COMMON_THREAD_POOL;
/**
* @author baigod
*/
@Service
@Slf4j
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='protocol'")
public class DownlinkGrpcService extends ProtocolInterfaceImplBase {
@Value("${service.protocol.rpc.port}")
private int rpcPort;
@Value("${service.protocol.rpc.boss}")
private int rpcBoss;
@Value("${service.protocol.rpc.worker}")
private int rpcWorker;
@Value("${service.protocol.rpc.so-rcvbuf}")
private int rpcNettySoRcvbuf;
@Value("${service.protocol.rpc.so-sndbuf}")
private int rpcNettySoSndbuf;
@Value("${service.protocol.rpc.no-delay}")
private boolean rpcNettyNoDelay;
@Value("${service.protocol.rpc.max-inbound-message-size}")
private int maxInboundMessageSize;
@Value("${service.protocol.rpc.max-concurrent-calls-per-connection}")
private int maxConcurrentCallsPerConnection;
@Value("${service.protocol.rpc.client-max-keep-alive-time-sec}")
private int clientMaxKeepAliveTimeSec;
@Resource
ProtocolSessionRegistryProvider protocolSessionRegistryProvider;
private Server server;
private static final ReentrantLock replyLock = new ReentrantLock();
@PostConstruct
public void init() throws Exception {
log.info("Initializing Protocol Downlink Grpc service!");
NettyServerBuilder builder = NettyServerBuilder.forPort(this.rpcPort)
.bossEventLoopGroup(new NioEventLoopGroup(this.rpcBoss))
.workerEventLoopGroup(new NioEventLoopGroup(this.rpcWorker))
.withOption(ChannelOption.SO_RCVBUF, rpcNettySoRcvbuf)
.withChildOption(ChannelOption.SO_SNDBUF, rpcNettySoSndbuf)
.withChildOption(ChannelOption.TCP_NODELAY, rpcNettyNoDelay)
.compressorRegistry(CompressorRegistry.getDefaultInstance())
.decompressorRegistry(DecompressorRegistry.getDefaultInstance())
.channelType(NioServerSocketChannel.class)
.permitKeepAliveTime(this.clientMaxKeepAliveTimeSec, TimeUnit.SECONDS)
.maxInboundMessageSize(maxInboundMessageSize)
.maxConcurrentCallsPerConnection(maxConcurrentCallsPerConnection)
.directExecutor()
.keepAliveTime(5, TimeUnit.MINUTES)
.keepAliveTimeout(10, TimeUnit.SECONDS)
.permitKeepAliveWithoutCalls(true)
.addService(this);
this.server = builder.build();
log.info("Going to start RPC server using port: {}", this.rpcPort);
try {
this.server.start();
} catch (Exception e) {
log.error("Failed to start RPC server!", e);
throw e;
}
log.info("Protocol Downlink Grpc service initialized!");
}
@PreDestroy
public void destroy() {
if (this.server != null) {
this.server.shutdownNow();
}
}
@Override
public StreamObserver<RequestMsg> onDownlink(StreamObserver<ResponseMsg> responseObserver) {
return new StreamObserver<>() {
@Override
public void onNext(RequestMsg requestMsg) {
TracerProto tracerProto = requestMsg.getTracer();
TracerContextUtil.newTracer(tracerProto.getId(), tracerProto.getOrigin(), tracerProto.getTs());
MDCUtils.recordTracer();
log.debug("通信层收到Grpc下行请求 {}", requestMsg);
if (requestMsg.hasConnectRequestMsg()) {
replyLock.lock();
try {
responseObserver.onNext(
ResponseMsg.newBuilder()
.setTracer(toTracerProto())
.setConnectResponseMsg(ConnectResponseMsg.newBuilder()
.setResponseCode(ConnectResponseCode.ACCEPTED)
.setErrorMsg("")
.build())
.build());
} finally {
replyLock.unlock();
}
}
if(requestMsg.hasDownlinkRequestMessage()){
DownlinkRequestMessage downlinkMsg = requestMsg.getDownlinkRequestMessage();
JCPP_COMMON_THREAD_POOL.execute(new TracerRunnable(() -> {
UUID protocolSessionId = new UUID(downlinkMsg.getSessionIdMSB(), downlinkMsg.getSessionIdLSB());
ProtocolSession protocolSession = protocolSessionRegistryProvider.get(protocolSessionId);
try {
if (protocolSession != null) {
protocolSession.onDownlink(downlinkMsg);
} else {
log.info("下发报文时Session未找到 sessionId: {}", protocolSessionId);
}
} catch (Exception e) {
log.warn("下发报文时处理失败 sessionId: {}", protocolSessionId, e);
}
}));
}
}
@Override
public void onError(Throwable t) {
log.error("Failed to deliver message from client!", t);
}
@Override
public void onCompleted() {
try {
responseObserver.onCompleted();
} catch (Exception e) {
log.error("onCompleted error ", e);
}
}
};
}
}

View File

@@ -20,4 +20,6 @@ public enum DownlinkCmdEnum {
REMOTE_START_CHARGING,
TRANSACTION_RECORD,
REMOTE_PARALLEL_START_CHARGING,
}

View File

@@ -9,7 +9,7 @@ import com.github.benmanes.caffeine.cache.Caffeine;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.forwarder.Forwarder;
import java.io.Closeable;
@@ -52,14 +52,14 @@ public abstract class ProtocolSession implements Closeable {
@Setter
private Forwarder forwarder;
public ProtocolSession(String protocolName) {
protected ProtocolSession(String protocolName) {
this.protocolName = protocolName;
this.pileCodeSet = new LinkedHashSet<>();
this.id = UUID.randomUUID();
this.lastActivityTime = LocalDateTime.now();
}
public abstract void onDownlink(DownlinkRestMessage downlinkMsg);
public abstract void onDownlink(DownlinkRequestMessage downlinkMsg);
public void close() {
close(SessionCloseReason.DESTRUCTION);

View File

@@ -4,10 +4,10 @@
*/
package sanbing.jcpp.protocol.domain;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
/**
* @author baigod
*/
public record SessionToHandlerMsg(DownlinkRestMessage downlinkMsg, ProtocolSession session) {
public record SessionToHandlerMsg(DownlinkRequestMessage downlinkMsg, ProtocolSession session) {
}

View File

@@ -25,10 +25,7 @@ import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
/**
* @author baigod
@@ -51,7 +48,7 @@ public abstract class Forwarder {
protected final boolean isMonolith;
protected QueueProducer<ProtoQueueMsg<UplinkQueueMessage>> producer;
public Forwarder(String protocolName, StatsFactory statsFactory, PartitionProvider partitionProvider, ServiceInfoProvider serviceInfoProvider) {
protected Forwarder(String protocolName, StatsFactory statsFactory, PartitionProvider partitionProvider, ServiceInfoProvider serviceInfoProvider) {
this.protocolName = protocolName;
this.partitionProvider = partitionProvider;
this.serviceInfoProvider = serviceInfoProvider;
@@ -66,12 +63,13 @@ public abstract class Forwarder {
public abstract void destroy();
protected void jcppForward(String topic, String key, UplinkQueueMessage msg, BiConsumer<Boolean, ObjectNode> consumer) {
forwarderMessagesStats.incrementTotal();
QueueMsgHeaders headers = new DefaultQueueMsgHeaders();
Tracer currentTracer = TracerContextUtil.getCurrentTracer();
headers.put(MSG_MD_PREFIX + JCPP_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId()));
headers.put(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin()));
headers.put(MSG_MD_PREFIX + MSG_MD_TS, ByteUtil.longToBytes(currentTracer.getTracerTs()));
headers.put(MSG_MD_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId()));
headers.put(MSG_MD_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin()));
headers.put(MSG_MD_TRACER_TS, ByteUtil.longToBytes(currentTracer.getTracerTs()));
TopicPartitionInfo tpi = partitionProvider.resolve(ServiceType.APP, topic, key);
producer.send(tpi, new ProtoQueueMsg<>(key, msg, headers), new QueueCallback() {
@@ -80,11 +78,13 @@ public abstract class Forwarder {
TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
MDCUtils.recordTracer();
log.trace("单体消息转发成功 key:{}", key);
if (consumer != null) {
consumer.accept(true, JacksonUtil.newObjectNode());
}
forwarderMessagesStats.incrementSuccessful();
}
@Override
@@ -92,6 +92,7 @@ public abstract class Forwarder {
TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
MDCUtils.recordTracer();
log.warn("单体消息转发异常", t);
if (consumer != null) {
@@ -99,6 +100,7 @@ public abstract class Forwarder {
objectNode.put(ERROR, t.getClass() + ": " + t.getMessage());
consumer.accept(true, objectNode);
}
forwarderMessagesStats.incrementFailed();
}
});
}

View File

@@ -36,10 +36,7 @@ import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_PREFIX;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.MSG_MD_TS;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ID;
import static sanbing.jcpp.infrastructure.util.trace.TracerContextUtil.JCPP_TRACER_ORIGIN;
import static sanbing.jcpp.infrastructure.queue.common.QueueConstants.*;
/**
* @author baigod
@@ -150,12 +147,13 @@ public class KafkaForwarder extends Forwarder {
}
private void kafkaForward(String topic, String key, UplinkQueueMessage msg, BiConsumer<Boolean, ObjectNode> consumer) throws InvalidProtocolBufferException {
forwarderMessagesStats.incrementTotal();
Headers headers = new RecordHeaders();
Tracer currentTracer = TracerContextUtil.getCurrentTracer();
headers.add(new RecordHeader(MSG_MD_PREFIX + JCPP_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId())));
headers.add(new RecordHeader(MSG_MD_PREFIX + JCPP_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin())));
headers.add(new RecordHeader(MSG_MD_PREFIX + MSG_MD_TS, ByteUtil.longToBytes(currentTracer.getTracerTs())));
headers.add(new RecordHeader(MSG_MD_TRACER_ID, ByteUtil.stringToBytes(currentTracer.getTraceId())));
headers.add(new RecordHeader(MSG_MD_TRACER_ORIGIN, ByteUtil.stringToBytes(currentTracer.getOrigin())));
headers.add(new RecordHeader(MSG_MD_TRACER_TS, ByteUtil.longToBytes(currentTracer.getTracerTs())));
if (kafkaCfg.getEncoder() == KafkaCfg.EncoderType.json) {
@@ -177,6 +175,7 @@ public class KafkaForwarder extends Forwarder {
private void logAndDoConsumer(BiConsumer<Boolean, ObjectNode> consumer, RecordMetadata metadata, Exception e, Tracer currentTracer) {
TracerContextUtil.newTracer(currentTracer.getTraceId(), currentTracer.getOrigin(), currentTracer.getTracerTs());
MDCUtils.recordTracer();
log.debug("Kafka 消息转发完成, success:{}", e == null);
if (consumer != null) {
@@ -196,6 +195,9 @@ public class KafkaForwarder extends Forwarder {
if (e != null) {
objectNode.put(ERROR, e.getClass() + ": " + e.getMessage());
forwarderMessagesStats.incrementFailed();
} else {
forwarderMessagesStats.incrementSuccessful();
}
consumer.accept(e == null, objectNode);

View File

@@ -19,7 +19,7 @@ import sanbing.jcpp.infrastructure.stats.MessagesStats;
import sanbing.jcpp.infrastructure.util.exception.DownlinkException;
import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.ProtocolMessageProcessor;
import sanbing.jcpp.protocol.domain.ListenerToHandlerMsg;
import sanbing.jcpp.protocol.domain.ProtocolUplinkMsg;
@@ -121,7 +121,7 @@ public class TcpChannelHandler<T> extends SimpleChannelInboundHandler<ProtocolUp
}
}
protected void onDownlink(DownlinkRestMessage downlinkMsg) throws DownlinkException {
protected void onDownlink(DownlinkRequestMessage downlinkMsg) throws DownlinkException {
protocolMessageProcessor.downlinkHandle(new SessionToHandlerMsg(downlinkMsg, tcpSession), downlinkMsgStats);
}

View File

@@ -9,7 +9,7 @@ import io.netty.channel.ChannelHandlerContext;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.Setter;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.domain.ProtocolSession;
import sanbing.jcpp.protocol.domain.SessionCloseReason;
import sanbing.jcpp.protocol.listener.tcp.enums.SequenceNumberLength;
@@ -32,7 +32,7 @@ public class TcpSession extends ProtocolSession {
private ChannelHandlerContext ctx;
private final Consumer<DownlinkRestMessage> sendDownlinkConsumer;
private final Consumer<DownlinkRequestMessage> sendDownlinkConsumer;
private final Consumer<ByteBuf> writeAndFlushConsumer;
@@ -64,7 +64,7 @@ public class TcpSession extends ProtocolSession {
}
public TcpSession(String protocolName,
Consumer<DownlinkRestMessage> sendDownlinkConsumer,
Consumer<DownlinkRequestMessage> sendDownlinkConsumer,
Consumer<ByteBuf> writeAndFlushConsumer) {
super(protocolName);
this.sendDownlinkConsumer = sendDownlinkConsumer;
@@ -72,7 +72,7 @@ public class TcpSession extends ProtocolSession {
}
@Override
public void onDownlink(DownlinkRestMessage downlinkMsg) {
public void onDownlink(DownlinkRequestMessage downlinkMsg) {
sendDownlinkConsumer.accept(downlinkMsg);
}

View File

@@ -34,10 +34,10 @@ public class DefaultProtocolSessionRegistryProvider implements ProtocolSessionRe
private static final int INIT_CACHE_LIMIT = 100_000;
private static final int MAXIMUM_SIZE = 1_000_000;
@Value("${service.protocols.sessions.default-inactivity-timeout-in-sec}")
@Value("${service.protocol.sessions.default-inactivity-timeout-in-sec}")
private int defaultInactivityTimeoutInSec;
@Value("${service.protocols.sessions.default-state-check-interval-in-sec}")
@Value("${service.protocol.sessions.default-state-check-interval-in-sec}")
private int defaultStateCheckIntervalInSec;
@Getter

View File

@@ -66,10 +66,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

View File

@@ -40,10 +40,23 @@ service:
type: "${SERVICE_TYPE:protocol}"
# 可自定义的服务ID如果不指定则默认为HOSTNAME
id: "${SERVICE_ID:}"
protocols:
protocol:
sessions:
default-inactivity-timeout-in-sec: "${PROTOCOLS_SESSIONS_DEFAULT_INACTIVITY_TIMEOUT_IN_SEC:600}"
default-state-check-interval-in-sec: "${PROTOCOLS_SESSIONS_DEFAULT_STATE_CHECK_INTERVAL_IN_SEC:60}"
rpc:
enabled: "${SERVICE_PROTOCOL_RPC_ENABLED:true}"
port: "${SERVICE_PROTOCOL_RPC_PORT:9090}"
boss: "${SERVICE_PROTOCOL_RPC_BOSS:4}"
worker: "${SERVICE_PROTOCOL_RPC_WORKER:64}"
so-rcvbuf: "${SERVICE_PROTOCOL_RPC_SO_RCVBUF:65535}"
so-sndbuf: "${SERVICE_PROTOCOL_RPC_SO_SNDBUF:65535}"
no-delay: "${SERVICE_PROTOCOL_RPC_NO_DELAY:true}"
user-thread-pool-size: "${SERVICE_PROTOCOL_RPC_USER_THREAD_POOL_SIZE:1024}"
max-inbound-message-size: "${SERVICE_PROTOCOL_RPC_MAX_INBOUND_MESSAGE_SIZE:33554432}"
max-concurrent-calls-per-connection: "${SERVICE_PROTOCOL_MAX_CONCURRENT_CALLS_PER_CONNECTION:4}"
client-max-keep-alive-time-sec: "${SERVICE_PROTOCOL_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:30}"
protocols:
yunkuaichongV150:
enabled: "${PROTOCOLS_YUNKUAICHONGV150_ENABLED:true}"
listener:
@@ -73,7 +86,7 @@ service:
# 以下配置只有在service.type为protocol时且jcpp-partition为false时才生效
bootstrap-servers: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_SERVERS:kafka:9092}"
acks: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_ACKS:1}"
# # 可选 protobuf推荐、json
# 可选 protobuf推荐、json
encoder: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_ENCODER:protobuf}"
retries: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_RETRIES:1}"
compression-type: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_COMPRESSION_TYPE:lz4}" # none, gzip, snappy, lz4, zstd
@@ -81,6 +94,43 @@ service:
linger-ms: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_KAFKA_LINGER_MS:0}"
buffer-memory: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_BUFFER_MEMORY:33554432}"
other-properties: "${PROTOCOLS_YUNKUAICHONGV150_FORWARD_QUEUE_KAFKA_OTHER_PROPERTIES:}"
yunkuaichongV160:
enabled: "${PROTOCOLS_YUNKUAICHONGV160_ENABLED:true}"
listener:
tcp:
bind-address: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BIND_ADDRESS:0.0.0.0}"
bind-port: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BIND_PORT:38002}"
boss-group-thread_count: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_BOSS_GROUP_THREADS:4}"
worker-group-thread-count: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_WORKER_GROUP_THREADS:16}"
so-keep-alive: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_KEEPALIVE:true}"
so-backlog: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_BACKLOG:128}"
so-rcvbuf: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_RCVBUF:65536}"
so-sndbuf: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_SO_SNDBUF:65536}"
nodelay: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_NODELAY:true}"
handler:
idle-timeout-seconds: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_HANDLER_IDLE_TIMEOUT_SECONDS:600}"
max_connections: "${PROTOCOLS_YUNKUAICHONGV160_LISTENER_TCP_HANDLER_MAX_CONNECTIONS:100000}"
# 默认为二进制类型的拆包器
# 可选JSON类型的拆包器 "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:JSON}"
# 可选纯文本类型的拆包器 "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:TEXT;maxFrameLength:128;stripDelimiter:true;messageSeparator:null;charsetName:UTF-8}"
configuration: "${PROTOCOLS_YUNKUAICHONGV160_NETTY_HANDLER_BINARY_CONFIGURATION:type:BINARY;decoder:sanbing.jcpp.protocol.listener.tcp.decoder.JCPPLengthFieldBasedFrameDecoder;byteOrder:LITTLE_ENDIAN;head:68;lengthFieldOffset:1;lengthFieldLength:1;lengthAdjustment:2;initialBytesToStrip:0}"
forwarder:
# 作为前置服务单独启时可选kafka、kafka-sharding未来计划扩展RocketMQ, GRpc、REST
type: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_TYPE:kafka}"
kafka:
topic: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_TOPIC:protocol_uplink}"
jcpp-partition: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_JCPP_PARTITION:true}" # 是否利用JCPP的分片框架
# 以下配置只有在service.type为protocol时且jcpp-partition为false时才生效
bootstrap-servers: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_SERVERS:kafka:9092}"
acks: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_ACKS:1}"
# 可选 protobuf推荐、json
encoder: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_ENCODER:protobuf}"
retries: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_RETRIES:1}"
compression-type: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_COMPRESSION_TYPE:lz4}" # none, gzip, snappy, lz4, zstd
batch-size: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_BATCH_SIZE:16384}"
linger-ms: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_KAFKA_LINGER_MS:0}"
buffer-memory: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_BUFFER_MEMORY:33554432}"
other-properties: "${PROTOCOLS_YUNKUAICHONGV160_FORWARD_QUEUE_KAFKA_OTHER_PROPERTIES:}"
# 应用程序服务注册中心配置
zk:
@@ -147,3 +197,4 @@ thread-pool:
hash_function_name: "${THREAD_POOL_SHARDING_HASH_FUNCTION_NAME:murmur3_128}" # murmur3_32, murmur3_128 or sha256
parallelism: "${THREAD_POOL_SHARDING_PARALLELISM:8}"
stats-print-interval-ms: "${THREAD_POOL_SHARDING_STATS_PRINT_INTERVAL_MS:10000}"

View File

@@ -21,7 +21,7 @@ import org.springframework.http.HttpStatus;
import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
import sanbing.jcpp.infrastructure.util.property.PropertyUtils;
import sanbing.jcpp.proto.gen.ProtocolProto;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.AbstractProtocolTestBase;
import sanbing.jcpp.protocol.domain.DownlinkCmdEnum;
import sanbing.jcpp.protocol.domain.ProtocolSession;
@@ -116,9 +116,9 @@ class DownlinkControllerIT extends AbstractProtocolTestBase {
UUID messageId = UUID.randomUUID();
UUID requestId = UUID.randomUUID();
// 创建 DownlinkRestMessage 实例
// 创建 DownlinkRequestMessage 实例
String pileCode = "20231212000010";
DownlinkRestMessage downlinkMsg = DownlinkRestMessage.newBuilder()
DownlinkRequestMessage downlinkMsg = DownlinkRequestMessage.newBuilder()
.setMessageIdMSB(messageId.getMostSignificantBits())
.setMessageIdLSB(messageId.getLeastSignificantBits())
.setSessionIdMSB(sessionId.getMostSignificantBits())
@@ -131,7 +131,7 @@ class DownlinkControllerIT extends AbstractProtocolTestBase {
.setRemoteStartChargingRequest(ProtocolProto.RemoteStartChargingRequest.newBuilder()
.setPileCode(pileCode)
.setGunCode("01")
.setLimitYuan(100)
.setLimitYuan("100")
.setTradeNo("12345678901234567890")
.build())
.build();

View File

@@ -12,7 +12,7 @@ import sanbing.jcpp.infrastructure.util.codec.BCDUtil;
import sanbing.jcpp.proto.gen.ProtocolProto;
import sanbing.jcpp.protocol.listener.tcp.TcpSession;
import sanbing.jcpp.protocol.listener.tcp.enums.SequenceNumberLength;
import sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum;
import sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum;
import java.math.BigDecimal;
import java.math.RoundingMode;
@@ -98,7 +98,7 @@ public class AbstractYunKuaiChongCmdExe {
}
protected byte[] encode(YunKuaiChongV150DownlinkCmdEnum downlinkCmd,
protected byte[] encode(YunKuaiChongDownlinkCmdEnum downlinkCmd,
int seqNo,
int encryptionFlag,
ByteBuf msgBody) {
@@ -120,7 +120,7 @@ public class AbstractYunKuaiChongCmdExe {
return toBytes(response);
}
protected void encodeAndWriteFlush(YunKuaiChongV150DownlinkCmdEnum downlinkCmd,
protected void encodeAndWriteFlush(YunKuaiChongDownlinkCmdEnum downlinkCmd,
int seqNo,
int encryptionFlag,
ByteBuf msgBody,
@@ -131,7 +131,7 @@ public class AbstractYunKuaiChongCmdExe {
tcpSession.writeAndFlush(Unpooled.copiedBuffer(encode));
}
protected void encodeAndWriteFlush(YunKuaiChongV150DownlinkCmdEnum downlinkCmd,
protected void encodeAndWriteFlush(YunKuaiChongDownlinkCmdEnum downlinkCmd,
ByteBuf msgBody,
TcpSession tcpSession) {

View File

@@ -10,7 +10,7 @@ import sanbing.jcpp.protocol.listener.tcp.TcpSession;
/**
* @author baigod
*/
public abstract class YunKuaiChongDownlinkCmdExe extends AbstractYunKuaiChongCmdExe{
public abstract class YunKuaiChongDownlinkCmdExe extends AbstractYunKuaiChongCmdExe {
public abstract void execute(TcpSession tcpSession, YunKuaiChongDwonlinkMessage yunKuaiChongDwonlinkMessage, ProtocolContext ctx);

View File

@@ -7,7 +7,7 @@ package sanbing.jcpp.protocol.yunkuaichong;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import java.io.Serializable;
import java.util.UUID;
@@ -32,7 +32,7 @@ public class YunKuaiChongDwonlinkMessage implements Serializable {
private int cmd;
// 消息体
private DownlinkRestMessage msg;
private DownlinkRequestMessage msg;
// 上行消息
private YunKuaiChongUplinkMessage requestData;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150;
package sanbing.jcpp.protocol.yunkuaichong;
import cn.hutool.core.util.ClassUtil;
import io.netty.buffer.ByteBuf;
@@ -10,19 +10,15 @@ import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import sanbing.jcpp.infrastructure.util.JCPPPair;
import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRestMessage;
import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
import sanbing.jcpp.protocol.ProtocolContext;
import sanbing.jcpp.protocol.ProtocolMessageProcessor;
import sanbing.jcpp.protocol.domain.ListenerToHandlerMsg;
import sanbing.jcpp.protocol.domain.SessionToHandlerMsg;
import sanbing.jcpp.protocol.forwarder.Forwarder;
import sanbing.jcpp.protocol.listener.tcp.TcpSession;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDownlinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum;
import sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum;
import java.lang.reflect.InvocationTargetException;
import java.util.Map;
@@ -33,17 +29,17 @@ import java.util.concurrent.ConcurrentHashMap;
import static sanbing.jcpp.infrastructure.util.codec.ByteUtil.checkCrcSum;
@Slf4j
public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProcessor {
private final Map<Byte, YunKuaiChongUplinkCmdExe> uplinkCmdExeMap = new ConcurrentHashMap<>();
private final Map<Byte, YunKuaiChongDownlinkCmdExe> downlinkCmdExeMap = new ConcurrentHashMap<>();
public class YunKuaiChongProtocolMessageProcessor extends ProtocolMessageProcessor {
private final Map<Integer, YunKuaiChongUplinkCmdExe> uplinkCmdExeMap = new ConcurrentHashMap<>();
private final Map<Integer, YunKuaiChongDownlinkCmdExe> downlinkCmdExeMap = new ConcurrentHashMap<>();
public YunKuaiChongV15ProtocolMessageProcessor(Forwarder forwarder, ProtocolContext protocolContext) {
public YunKuaiChongProtocolMessageProcessor(Forwarder forwarder, ProtocolContext protocolContext) {
super(forwarder, protocolContext);
Set<Class<?>> cmdClasses = ClassUtil.scanPackageByAnnotation(ClassUtil.getPackage(this.getClass()), YunKuaiChongCmd.class);
cmdClasses.stream().filter(YunKuaiChongUplinkCmdExe.class::isAssignableFrom)
.forEach(clazz -> {
byte cmd = clazz.getAnnotation(YunKuaiChongCmd.class).value();
int cmd = clazz.getAnnotation(YunKuaiChongCmd.class).value();
try {
YunKuaiChongUplinkCmdExe yunKuaiChongUplinkCmdExe = (YunKuaiChongUplinkCmdExe) clazz.getDeclaredConstructor().newInstance();
uplinkCmdExeMap.put(cmd, yunKuaiChongUplinkCmdExe);
@@ -57,7 +53,7 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
cmdClasses.stream().filter(YunKuaiChongDownlinkCmdExe.class::isAssignableFrom)
.forEach(clazz -> {
byte cmd = clazz.getAnnotation(YunKuaiChongCmd.class).value();
int cmd = clazz.getAnnotation(YunKuaiChongCmd.class).value();
try {
YunKuaiChongDownlinkCmdExe yunKuaiChongDownlinkCmdExe = (YunKuaiChongDownlinkCmdExe) clazz.getDeclaredConstructor().newInstance();
downlinkCmdExeMap.put(cmd, yunKuaiChongDownlinkCmdExe);
@@ -138,15 +134,15 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
JCPPPair<Boolean, Integer> checkResult = checkCrcSum(checkData, checkSum);
if (!checkResult.getFirst()) {
if (Boolean.FALSE.equals(checkResult.getFirst())) {
csTemp.writeBytes(byCheckSum);
checkSum = csTemp.readUnsignedShortLE();
checkResult = checkCrcSum(checkData, checkSum);
log.debug("云快充V1.5检验和 第二次检查: checkResult:{}, checkSum:{}", checkResult, checkSum);
log.debug("云快充检验和 第二次检查: checkResult:{}, checkSum:{}", checkResult, checkSum);
}
if (!checkResult.getFirst()) {
log.info("云快充V1.5检验和不一致两次不通过 不处理! CMD{},校验域:{},正确校验和:{}", frameType, checkSum, checkResult.getSecond());
if (Boolean.FALSE.equals(checkResult.getFirst())) {
log.info("云快充检验和不一致两次不通过 不处理! CMD{},校验域:{},正确校验和:{}", frameType, checkSum, checkResult.getSecond());
return;
}
@@ -167,9 +163,9 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
public void downlinkHandle(SessionToHandlerMsg sessionToHandlerMsg) {
TcpSession session = (TcpSession) sessionToHandlerMsg.session();
DownlinkRestMessage protocolDownlinkMsg = sessionToHandlerMsg.downlinkMsg();
DownlinkRequestMessage protocolDownlinkMsg = sessionToHandlerMsg.downlinkMsg();
int cmd = YunKuaiChongV150DownlinkCmdEnum.valueOf(protocolDownlinkMsg.getDownlinkCmd()).getCmd();
int cmd = YunKuaiChongDownlinkCmdEnum.valueOf(protocolDownlinkMsg.getDownlinkCmd()).getCmd();
YunKuaiChongDwonlinkMessage message = new YunKuaiChongDwonlinkMessage();
message.setId(new UUID(protocolDownlinkMsg.getMessageIdMSB(), protocolDownlinkMsg.getMessageIdLSB()));
@@ -188,11 +184,11 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
}
private void exeCmd(YunKuaiChongUplinkMessage message, TcpSession session) {
YunKuaiChongUplinkCmdExe uplinkCmdExe = uplinkCmdExeMap.get((byte) message.getCmd());
YunKuaiChongUplinkCmdExe uplinkCmdExe = uplinkCmdExeMap.get(message.getCmd());
if (uplinkCmdExe == null) {
log.info("[{}] 云快充V1.5协议接收到未知的上行指令 {}", session, message.getCmd());
log.info("[{}] 云快充协议接收到未知的上行指令 {}", session, message.getCmd());
return;
}
@@ -201,11 +197,11 @@ public class YunKuaiChongV15ProtocolMessageProcessor extends ProtocolMessageProc
}
private void exeCmd(YunKuaiChongDwonlinkMessage message, TcpSession session) {
YunKuaiChongDownlinkCmdExe downlinkCmdExe = downlinkCmdExeMap.get((byte) message.getCmd());
YunKuaiChongDownlinkCmdExe downlinkCmdExe = downlinkCmdExeMap.get(message.getCmd());
if (downlinkCmdExe == null) {
log.info("[{}] 云快充V1.5协议接收到未知的下行指令 {}", session, message.getCmd());
log.info("[{}] 云快充协议接收到未知的下行指令 {}", session, message.getCmd());
return;
}

View File

@@ -15,7 +15,7 @@ import sanbing.jcpp.protocol.listener.tcp.TcpSession;
* @author baigod
*/
@Slf4j
public abstract class YunKuaiChongUplinkCmdExe extends AbstractYunKuaiChongCmdExe{
public abstract class YunKuaiChongUplinkCmdExe extends AbstractYunKuaiChongCmdExe {
public abstract void execute(TcpSession tcpSession, YunKuaiChongUplinkMessage yunKuaiChongUplinkMessage, ProtocolContext ctx);

View File

@@ -14,6 +14,6 @@ import java.lang.annotation.*;
@Documented
public @interface YunKuaiChongCmd {
byte value();
int value();
}

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.enums;
package sanbing.jcpp.protocol.yunkuaichong.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
@@ -12,7 +12,7 @@ import lombok.Getter;
*/
@AllArgsConstructor
@Getter
public enum YunKuaiChongV150DownlinkCmdEnum {
public enum YunKuaiChongDownlinkCmdEnum {
LOGIN_ACK(0x02),
@@ -30,9 +30,10 @@ public enum YunKuaiChongV150DownlinkCmdEnum {
REMOTE_STOP_CHARGING(0x36),
TRANSACTION_RECORD(0x40)
;
TRANSACTION_RECORD(0x40),
private int cmd;
REMOTE_PARALLEL_START_CHARGING(0xA4);
private final Integer cmd;
}

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import cn.hutool.core.util.HexUtil;
import com.fasterxml.jackson.databind.node.ObjectNode;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;
@@ -18,7 +18,7 @@ import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.HEARTBEAT;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.HEARTBEAT;
/**
* 云快充1.5.0 充电桩心跳包
@@ -54,7 +54,9 @@ public class YunKuaiChongV150HeartbeatULCmd extends YunKuaiChongUplinkCmdExe {
.setPileCode(pileCode)
.setRemoteAddress(tcpSession.getAddress().toString())
.setNodeId(ctx.getServiceInfoProvider().getServiceId())
.setNodeWebapiIpPort(ctx.getServiceInfoProvider().getServiceWebapiEndpoint())
.setNodeHostAddress(ctx.getServiceInfoProvider().getHostAddress())
.setNodeRestPort(ctx.getServiceInfoProvider().getRestPort())
.setNodeGrpcPort(ctx.getServiceInfoProvider().getGrpcPort())
.setAdditionalInfo(additionalInfo.toString())
.build();
UplinkQueueMessage uplinkQueueMessage = uplinkMessageBuilder(heartBeatRequest.getPileCode(), tcpSession, yunKuaiChongUplinkMessage)

View File

@@ -2,8 +2,9 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import cn.hutool.core.util.RandomUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
@@ -28,8 +29,8 @@ import static sanbing.jcpp.infrastructure.util.config.ThreadPoolConfiguration.PR
import static sanbing.jcpp.protocol.domain.SessionCloseReason.MANUALLY;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.FAILURE_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.SUCCESS_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.LOGIN_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.SYNC_TIME;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.LOGIN_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.SYNC_TIME;
/**
* 云快充1.5.0登录认证应答
@@ -95,7 +96,7 @@ public class YunKuaiChongV150LoginAckDLCmd extends YunKuaiChongDownlinkCmdExe {
log.info("{} 云快充1.5.0开始注册定时对时任务", tcpSession);
return PROTOCOL_SESSION_SCHEDULED.scheduleAtFixedRate(() ->
syncTime(tcpSession, pileCodeBytes, requestData),
0, 8, TimeUnit.HOURS);
0, RandomUtil.randomInt(420, 480), TimeUnit.MINUTES);
}
);
}

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;
@@ -66,7 +66,9 @@ public class YunKuaiChongV150LoginULCmd extends YunKuaiChongUplinkCmdExe {
.setCredential(pileCode)
.setRemoteAddress(tcpSession.getAddress().toString())
.setNodeId(ctx.getServiceInfoProvider().getServiceId())
.setNodeWebapiIpPort(ctx.getServiceInfoProvider().getServiceWebapiEndpoint())
.setNodeHostAddress(ctx.getServiceInfoProvider().getHostAddress())
.setNodeRestPort(ctx.getServiceInfoProvider().getRestPort())
.setNodeGrpcPort(ctx.getServiceInfoProvider().getGrpcPort())
.setAdditionalInfo(additionalInfo.toString())
.build();
UplinkQueueMessage uplinkQueueMessage = uplinkMessageBuilder(loginRequest.getPileCode(), tcpSession, yunKuaiChongUplinkMessage)

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -23,7 +23,7 @@ import java.util.List;
import java.util.Map;
import static sanbing.jcpp.proto.gen.ProtocolProto.PricingModelFlag.*;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.QUERY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.QUERY_PRICING_ACK;
/**
* 计费模型请求应答

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -16,7 +16,9 @@ import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDownlinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.REMOTE_START_CHARGING;
import java.math.BigDecimal;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.REMOTE_START_CHARGING;
/**
* 云快充1.5.0 运营平台远程控制启机
@@ -39,7 +41,7 @@ public class YunKuaiChongV150RemoteStartDLCmd extends YunKuaiChongDownlinkCmdExe
String pileCode = remoteStartChargingRequest.getPileCode();
String gunCode = remoteStartChargingRequest.getGunCode();
String tradeNo = remoteStartChargingRequest.getTradeNo();
int limitYuan = remoteStartChargingRequest.getLimitYuan();
String limitYuan = remoteStartChargingRequest.getLimitYuan();
byte[] cardNo = encodeCardNo(tradeNo);
@@ -55,7 +57,7 @@ public class YunKuaiChongV150RemoteStartDLCmd extends YunKuaiChongDownlinkCmdExe
// 物理卡号
msgBody.writeBytes(cardNo);
// 账户余额
msgBody.writeIntLE(limitYuan);
msgBody.writeIntLE(new BigDecimal(limitYuan).intValue());
encodeAndWriteFlush(REMOTE_START_CHARGING,
msgBody,

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -14,7 +14,7 @@ import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDownlinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.REMOTE_START_CHARGING;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.REMOTE_START_CHARGING;
/**
* 云快充1.5.0 运营平台远程停机

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -16,7 +16,7 @@ import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.QUERY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.QUERY_PRICING_ACK;
/**
* 云快充1.5.0 计费模型应答

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -23,8 +23,8 @@ import java.util.List;
import java.util.Map;
import static sanbing.jcpp.proto.gen.ProtocolProto.PricingModelFlag.*;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.QUERY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.SET_PRICING;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.QUERY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.SET_PRICING;
/**
* 云快充1.5.0 计费模型设置

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -18,7 +18,7 @@ import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.FAILURE_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.SUCCESS_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.VERIFY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.VERIFY_PRICING_ACK;
/**
* 云快充1.5.0 交易记录确认

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -20,7 +20,7 @@ import java.util.Arrays;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.FAILURE_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage.SUCCESS_BYTE;
import static sanbing.jcpp.protocol.yunkuaichong.v150.enums.YunKuaiChongV150DownlinkCmdEnum.VERIFY_PRICING_ACK;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.VERIFY_PRICING_ACK;
/**
* 云快充1.5.0计费模型验证请求应答

View File

@@ -2,7 +2,7 @@
* 抖音关注程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v150.cmd;
package sanbing.jcpp.protocol.yunkuaichong.v150;
import com.fasterxml.jackson.databind.node.ObjectNode;

View File

@@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j;
import sanbing.jcpp.infrastructure.util.annotation.ProtocolComponent;
import sanbing.jcpp.protocol.ProtocolBootstrap;
import sanbing.jcpp.protocol.ProtocolMessageProcessor;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongProtocolMessageProcessor;
import static sanbing.jcpp.protocol.yunkuaichong.v150.YunkuaichongV150ProtocolBootstrap.PROTOCOL_NAME;
@@ -38,7 +39,7 @@ public class YunkuaichongV150ProtocolBootstrap extends ProtocolBootstrap {
@Override
protected ProtocolMessageProcessor messageProcessor() {
return new YunKuaiChongV15ProtocolMessageProcessor(forwarder, protocolContext);
return new YunKuaiChongProtocolMessageProcessor(forwarder, protocolContext);
}

View File

@@ -0,0 +1,46 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v160;
import lombok.extern.slf4j.Slf4j;
import sanbing.jcpp.infrastructure.util.annotation.ProtocolComponent;
import sanbing.jcpp.protocol.ProtocolBootstrap;
import sanbing.jcpp.protocol.ProtocolMessageProcessor;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongProtocolMessageProcessor;
import static sanbing.jcpp.protocol.yunkuaichong.v160.YunkuaichongV160ProtocolBootstrap.PROTOCOL_NAME;
/**
* @author baigod
*/
@ProtocolComponent(PROTOCOL_NAME)
@Slf4j
public class YunkuaichongV160ProtocolBootstrap extends ProtocolBootstrap {
public static final String PROTOCOL_NAME = "yunkuaichongV160";
@Override
protected String getProtocolName() {
return PROTOCOL_NAME;
}
@Override
protected void _init() {
// do nothing
}
@Override
protected void _destroy() {
// do nothing
}
@Override
protected ProtocolMessageProcessor messageProcessor() {
return new YunKuaiChongProtocolMessageProcessor(forwarder, protocolContext);
}
}

View File

@@ -0,0 +1,81 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v160.cmd;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import sanbing.jcpp.infrastructure.util.codec.BCDUtil;
import sanbing.jcpp.proto.gen.ProtocolProto;
import sanbing.jcpp.protocol.ProtocolContext;
import sanbing.jcpp.protocol.listener.tcp.TcpSession;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDownlinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongDwonlinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import static sanbing.jcpp.protocol.yunkuaichong.enums.YunKuaiChongDownlinkCmdEnum.REMOTE_START_CHARGING;
/**
* 云快充1.6.0 运营平台远程控制并充启机
*
* @author baigod
*/
@Slf4j
@YunKuaiChongCmd(0xA4)
public class YunKuaiChongV160RemoteParallelStartDLCmd extends YunKuaiChongDownlinkCmdExe {
static final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
@Override
public void execute(TcpSession tcpSession, YunKuaiChongDwonlinkMessage yunKuaiChongDwonlinkMessage, ProtocolContext ctx) {
log.info("{} 云快充1.6.0运营平台远程控制并充启机", tcpSession);
if (!yunKuaiChongDwonlinkMessage.getMsg().hasRemoteStartChargingRequest()) {
return;
}
ProtocolProto.RemoteStartChargingRequest remoteStartChargingRequest = yunKuaiChongDwonlinkMessage.getMsg().getRemoteStartChargingRequest();
String pileCode = remoteStartChargingRequest.getPileCode();
String gunCode = remoteStartChargingRequest.getGunCode();
String tradeNo = remoteStartChargingRequest.getTradeNo();
String limitYuan = remoteStartChargingRequest.getLimitYuan();
byte[] cardNo = encodeCardNo(tradeNo);
ByteBuf msgBody = Unpooled.buffer(44);
// 交易流水号
msgBody.writeBytes(encodeTradeNo(tradeNo));
// 桩编码
msgBody.writeBytes(encodePileCode(pileCode));
// 枪号
msgBody.writeBytes(encodeGunCode(gunCode));
// 逻辑卡号 BCD码
msgBody.writeBytes(cardNo);
// 物理卡号
msgBody.writeBytes(cardNo);
// 账户余额
msgBody.writeIntLE(new BigDecimal(limitYuan).intValue());
// 并充序号
msgBody.writeBytes(BCDUtil.toBytes(LocalDateTime.now().format(dateTimeFormatter)));
encodeAndWriteFlush(REMOTE_START_CHARGING,
msgBody,
tcpSession);
}
/**
* 用交易流水号做卡号
*/
private static byte[] encodeCardNo(String tradeNo) {
tradeNo = StringUtils.right(tradeNo, 16);
tradeNo = StringUtils.leftPad(tradeNo, 16, '0');
return BCDUtil.toBytes(tradeNo);
}
}

View File

@@ -0,0 +1,104 @@
/**
* 抖音关注:程序员三丙
* 知识星球https://t.zsxq.com/j9b21
*/
package sanbing.jcpp.protocol.yunkuaichong.v160.cmd;
import com.fasterxml.jackson.databind.node.ObjectNode;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import sanbing.jcpp.infrastructure.util.codec.BCDUtil;
import sanbing.jcpp.infrastructure.util.jackson.JacksonUtil;
import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
import sanbing.jcpp.proto.gen.ProtocolProto.RemoteStartChargingResponse;
import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
import sanbing.jcpp.protocol.ProtocolContext;
import sanbing.jcpp.protocol.listener.tcp.TcpSession;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkCmdExe;
import sanbing.jcpp.protocol.yunkuaichong.YunKuaiChongUplinkMessage;
import sanbing.jcpp.protocol.yunkuaichong.annotation.YunKuaiChongCmd;
/**
* 云快充1.6.0 远程并充启机命令回复
*
* @author baigod
*/
@Slf4j
@YunKuaiChongCmd(0xA3)
public class YunKuaiChongV160RemoteParallelStartResultULCmd extends YunKuaiChongUplinkCmdExe {
@Override
public void execute(TcpSession tcpSession, YunKuaiChongUplinkMessage yunKuaiChongUplinkMessage, ProtocolContext ctx) {
log.info("{} 云快充1.6.远程并充启机命令回复", tcpSession);
ByteBuf byteBuf = Unpooled.copiedBuffer(yunKuaiChongUplinkMessage.getMsgBody());
// 从Tracer总获取当前时间
long ts = TracerContextUtil.getCurrentTracer().getTracerTs();
ObjectNode additionalInfo = JacksonUtil.newObjectNode();
// 1.交易流水号
byte[] tradeNoBytes = new byte[16];
byteBuf.readBytes(tradeNoBytes);
String tradeNo = decodeTradeNo(tradeNoBytes);
// 2.桩编号
byte[] pileCodeBytes = new byte[7];
byteBuf.readBytes(pileCodeBytes);
String pileCode = BCDUtil.toString(pileCodeBytes);
// 3.抢号
byte gunCodeByte = byteBuf.readByte();
String gunCode = BCDUtil.toString(gunCodeByte);
// 4.命令执行结果 0x00失败 0x01成功
boolean isSuccess = (byteBuf.readByte() == 0x01);
// 5.失败原因 0无 1设备编码不匹配 2枪已在充电 3设备故障 4设备离线 5未插枪
byte failReasonByte = byteBuf.readByte();
String failReason = mapFailCode(failReasonByte);
// 6.主辅枪标记 0x00 主枪 0x01辅枪
byte gunFlagByte = byteBuf.readByte();
String gunFlag = BCDUtil.toString(gunFlagByte);
additionalInfo.put("主辅枪标记", gunFlag);
// 7.并充序号0xA4下发的并充序号
byte[] parallelSeqNoBytes = new byte[6];
byteBuf.readBytes(parallelSeqNoBytes);
String parallelSeqNo = BCDUtil.toString(parallelSeqNoBytes);
additionalInfo.put("并充序号", parallelSeqNo);
RemoteStartChargingResponse remoteStartChargingResponse = RemoteStartChargingResponse.newBuilder()
.setTs(ts)
.setPileCode(pileCode)
.setGunCode(gunCode)
.setTradeNo(tradeNo)
.setSuccess(isSuccess)
.setFailReason(failReason)
.setAdditionalInfo(additionalInfo.toString())
.build();
// 转发到后端
UplinkQueueMessage uplinkQueueMessage = uplinkMessageBuilder(pileCode, tcpSession, yunKuaiChongUplinkMessage)
.setRemoteStartChargingResponse(remoteStartChargingResponse)
.build();
tcpSession.getForwarder().sendMessage(uplinkQueueMessage);
}
public static String mapFailCode(byte failCode) {
return switch (failCode) {
case 0x00 -> "";
case 0x01 -> "设备编号不匹配";
case 0x02 -> "枪已在充电";
case 0x03 -> "设备故障";
case 0x04 -> "设备离线";
case 0x05 -> "未插枪";
case 0x33 -> "充电失败"; // 充电失败或其他相关信息
case 0x34 -> "待启充"; // 示例,处理收到充电命令
default -> "未知错误代码";
};
}
}

33
pom.xml
View File

@@ -40,10 +40,10 @@
<skip.integration.tests/>
<os-maven-plugin.version>1.7.0</os-maven-plugin.version>
<disruptor.version>3.4.4</disruptor.version>
<protobuf.version>3.21.9</protobuf.version>
<jakarta.el.version>4.0.2</jakarta.el.version>
<grpc.version>1.56.1</grpc.version>
<protobuf-maven-plugin.version>0.5.1</protobuf-maven-plugin.version>
<protobuf.version>4.28.2</protobuf.version>
<grpc.version>1.68.0</grpc.version>
<protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
<maven-compiler-plugin.version>3.13.0</maven-compiler-plugin.version>
<hutool-all.version>5.8.32</hutool-all.version>
<mybatis-plus-boot-starter.version>3.5.7</mybatis-plus-boot-starter.version>
@@ -51,6 +51,7 @@
<oshi-core.version>6.6.2</oshi-core.version>
<zookeeper.version>3.9.2</zookeeper.version>
<xnio-api.version>3.8.16.Final</xnio-api.version>
<javax.annotation-api.version>1.3.2</javax.annotation-api.version>
</properties>
<profiles>
@@ -180,6 +181,11 @@
<artifactId>xnio-api</artifactId>
<version>${xnio-api.version}</version>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<version>${javax.annotation-api.version}</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
@@ -190,6 +196,21 @@
<artifactId>protobuf-java-util</artifactId>
<version>${protobuf.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
<version>${grpc.version}</version>
</dependency>
<dependency>
<groupId>org.glassfish</groupId>
<artifactId>jakarta.el</artifactId>
@@ -243,7 +264,7 @@
<artifactId>maven-compiler-plugin</artifactId>
<version>${maven-compiler-plugin.version}</version>
<configuration>
<release>21</release>
<release>${java.version}</release>
<compilerArgs>
<arg>-Xlint:deprecation</arg>
<arg>-Xlint:removal</arg>
@@ -361,6 +382,10 @@
</plugins>
</pluginManagement>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>