Commit 02585823fcf3e311998ef701eb88204f86dae2c7
1 parent
7644aa43
Properly hanlde gRPC session timeout
Showing
3 changed files
with
13 additions
and
3 deletions
... | ... | @@ -19,7 +19,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; |
19 | 19 | import com.google.common.io.Resources; |
20 | 20 | import com.google.common.util.concurrent.FutureCallback; |
21 | 21 | import io.grpc.Server; |
22 | -import io.grpc.ServerBuilder; | |
22 | +import io.grpc.netty.NettyServerBuilder; | |
23 | 23 | import io.grpc.stub.StreamObserver; |
24 | 24 | import lombok.extern.slf4j.Slf4j; |
25 | 25 | import org.springframework.beans.factory.annotation.Autowired; |
... | ... | @@ -49,6 +49,7 @@ import java.util.Map; |
49 | 49 | import java.util.concurrent.ConcurrentHashMap; |
50 | 50 | import java.util.concurrent.ExecutorService; |
51 | 51 | import java.util.concurrent.Executors; |
52 | +import java.util.concurrent.TimeUnit; | |
52 | 53 | |
53 | 54 | @Service |
54 | 55 | @Slf4j |
... | ... | @@ -68,6 +69,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i |
68 | 69 | private String privateKeyResource; |
69 | 70 | @Value("${edges.state.persistToTelemetry:false}") |
70 | 71 | private boolean persistToTelemetry; |
72 | + @Value("${edges.rpc.client_max_keep_alive_time_sec}") | |
73 | + private int clientMaxKeepAliveTimeSec; | |
71 | 74 | |
72 | 75 | @Autowired |
73 | 76 | private EdgeContextComponent ctx; |
... | ... | @@ -82,7 +85,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i |
82 | 85 | @PostConstruct |
83 | 86 | public void init() { |
84 | 87 | log.info("Initializing Edge RPC service!"); |
85 | - ServerBuilder builder = ServerBuilder.forPort(rpcPort).addService(this); | |
88 | + NettyServerBuilder builder = NettyServerBuilder.forPort(rpcPort) | |
89 | + .permitKeepAliveTime(clientMaxKeepAliveTimeSec, TimeUnit.SECONDS) | |
90 | + .addService(this); | |
86 | 91 | if (sslEnabled) { |
87 | 92 | try { |
88 | 93 | File certFile = new File(Resources.getResource(certFileResource).toURI()); | ... | ... |
... | ... | @@ -590,6 +590,7 @@ edges: |
590 | 590 | rpc: |
591 | 591 | enabled: "${EDGES_RPC_ENABLED:false}" |
592 | 592 | port: "${EDGES_RPC_PORT:7070}" |
593 | + client_max_keep_alive_time_sec: "${EDGES_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:300}" | |
593 | 594 | ssl: |
594 | 595 | # Enable/disable SSL support |
595 | 596 | enabled: "${EDGES_RPC_SSL_ENABLED:false}" | ... | ... |
... | ... | @@ -55,6 +55,8 @@ public class EdgeGrpcClient implements EdgeRpcClient { |
55 | 55 | private int rpcPort; |
56 | 56 | @Value("${cloud.rpc.timeout}") |
57 | 57 | private int timeoutSecs; |
58 | + @Value("${cloud.rpc.keep_alive_time_sec}") | |
59 | + private int keepAliveTimeSec; | |
58 | 60 | @Value("${cloud.rpc.ssl.enabled}") |
59 | 61 | private boolean sslEnabled; |
60 | 62 | @Value("${cloud.rpc.ssl.cert}") |
... | ... | @@ -73,7 +75,9 @@ public class EdgeGrpcClient implements EdgeRpcClient { |
73 | 75 | Consumer<EdgeConfiguration> onEdgeUpdate, |
74 | 76 | Consumer<DownlinkMsg> onDownlink, |
75 | 77 | Consumer<Exception> onError) { |
76 | - NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort).usePlaintext(); | |
78 | + NettyChannelBuilder builder = NettyChannelBuilder.forAddress(rpcHost, rpcPort) | |
79 | + .keepAliveTime(keepAliveTimeSec, TimeUnit.SECONDS) | |
80 | + .usePlaintext(); | |
77 | 81 | if (sslEnabled) { |
78 | 82 | try { |
79 | 83 | builder.sslContext(GrpcSslContexts.forClient().trustManager(new File(Resources.getResource(certResource).toURI())).build()); | ... | ... |