neton.client.registry.HeartBeat source code

1 module neton.client.registry.HeartBeat;
2 
3 import hunt.logging;
4 import grpc;
5 import neton.protocol.neton;
6 import neton.protocol.neton;
7 import neton.protocol.netonrpc;
8 
9 import hunt.concurrency.ScheduledThreadPoolExecutor;
10 import hunt.concurrency.ThreadPoolExecutor;
11 import hunt.concurrency.Executors;
12 import hunt.concurrency.ExecutorService;
13 import hunt.concurrency.Future;
14 import hunt.util.Common;
15 import core.time;
16 import std.stdio;
17 
18 class HeartBeat
19 {
20     private
21     {
22         LeaseClient _client;
23         long _leaseID;
24         ScheduledThreadPoolExecutor _executor;
25     }
26 
27     this(Channel channel)
28     {
29         _client = new LeaseClient(channel);
30         _executor = cast(ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(1);
31     }
32 
33     bool grant(long ttl, ref long ID)
34     {
35         assert(ttl > 1);
36 
37         LeaseGrantRequest request = new LeaseGrantRequest();
38         request.TTL = ttl;
39         request.ID = ID;
40 
41         LeaseGrantResponse response = _client.LeaseGrant(request);
42         if (response is null)
43         {
44             logError("LeaseGrant error : ",ID);
45             return false;
46         }
47         ID = response.ID;
48         _leaseID = ID;
49         return true;
50     }
51 
52     void start()
53     {
54         synchronized (this)
55         {
56             LeaseKeepAliveRequest request = new LeaseKeepAliveRequest();
57             request.ID = _leaseID;
58             auto stream = _client.LeaseKeepAlive();
59             stream.write(request);
60             LeaseKeepAliveResponse response;
61             stream.read(response);
62 
63             _executor.scheduleAtFixedRate(new HeartBeatTask(stream,request),
64                     seconds(response.TTL/2 + 1), seconds(response.TTL - 1));
65         }
66     }
67 
68     bool stop()
69     {
70         synchronized (this)
71         {
72             version(HUNT_DEBUG)logInfo("------ stop heartBeat : ",_leaseID);
73             _executor.shutdown();
74             LeaseRevokeRequest request = new LeaseRevokeRequest();
75             request.ID = _leaseID;
76             LeaseRevokeResponse response = _client.LeaseRevoke(request);
77             if (response is null)
78             {
79                 logError("LeaseRevoke error : ",_leaseID);
80                 return false;
81             }
82             return true;
83         }
84     }
85 }
86 
87 class HeartBeatTask : Runnable
88 {
89     private
90     {
91         LeaseKeepAliveRequest _request;
92         ClientReaderWriter!(LeaseKeepAliveResponse,LeaseKeepAliveRequest) _stream;
93     }
94 
95     this(ClientReaderWriter!(LeaseKeepAliveResponse,LeaseKeepAliveRequest) stream, LeaseKeepAliveRequest req)
96     {
97         _stream = stream;  
98         _request = req;
99     }
100 
101     void run()
102     {
103         version(HUNT_DEBUG)logInfo("send heart beat : ",_request.ID);
104         LeaseKeepAliveResponse response;
105         _stream.write(_request);
106         _stream.read(response);
107     }
108 
109 }