neton.client.registry.RegistryService source code

1 module neton.client.registry.RegistryService;
2 
3 import neton.protocol.neton;
4 import neton.protocol.neton;
5 import neton.protocol.netonrpc;
6 
7 import grpc;
8 import hunt.logging;
9 import hunt.util.Serialize;
10 import std.conv;
11 import std.json;
12 
13 import neton.client.registry.Instance;
14 import neton.client.Listener;
15 import neton.client.WatchFactory;
16 import neton.client.Watcher;
17 import neton.client.registry.HeartBeat;
18 
19 enum HEART_TTL = 60; ///seconds
20 
21 class RegistryService
22 {
23     private
24     {
25         RegistryClient _client;
26         Watcher _watcher;
27         HeartBeat[long] _heartBeatMap;
28         Channel _channel;
29     }
30 
31     public this(Channel chanel)
32     {
33         _channel = chanel;
34         _client = new RegistryClient(chanel);
35         _watcher = WatchFactory.createRegistryWatcher(chanel, "/service");
36     }
37 
38     ~this()
39     {
40         if(_heartBeatMap.length > 0)
41         {
42             foreach(long leaseId , HeartBeat hb; _heartBeatMap) {
43                 hb.stop();
44             }
45             _heartBeatMap.clear();
46         }
47     }
48 
49     bool registerInstance(string serviceName, string ip, short port)
50     {
51         logInfo("registerInstance : ",serviceName);
52 
53         try
54         {
55             string regKey = serviceName ~ "/" ~ ip ~ ":" ~ to!string(port);
56             long leaseID = hashOf(regKey);
57             if (leaseID in _heartBeatMap)
58             {
59                 logErrorf("%s have already registered!", regKey);
60                 return false;
61             }
62             auto heartBeat = new HeartBeat(_channel);
63             heartBeat.grant(HEART_TTL, leaseID);
64 
65             Instance instant = {serviceName, ip, port};
66 
67             PutRequest request = new PutRequest();
68             request.key = cast(ubyte[])(regKey);
69             request.value = cast(ubyte[])(toJson(instant).toString);
70             request.lease = leaseID;
71             assert(request.key.length != 0);
72 
73             PutResponse response = _client.Put(request);
74             if (response is null)
75             {
76                 logError("response is null");
77                 return false;
78             }
79             _heartBeatMap[leaseID] = heartBeat;
80             heartBeat.start();
81 
82             return true;
83         }
84         catch (Throwable e)
85         {
86             logError("RegistryService : ", e.msg);
87             return false;
88         }
89     }
90 
91     bool deregisterInstance(string serviceName, string ip, short port)
92     {
93         logInfo("deregisterInstance : ",serviceName);
94         try
95         {
96             string regKey = serviceName ~ "/" ~ ip ~ ":" ~ to!string(port);
97             long leaseID = hashOf(regKey);
98 
99             DeleteRangeRequest request = new DeleteRangeRequest();
100             request.key = cast(ubyte[])(regKey);
101             assert(request.key.length != 0);
102 
103             DeleteRangeResponse response = _client.DeleteRange(request);
104             if (response is null)
105             {
106                 logError("response is null");
107                 return false;
108             }
109             if (leaseID in _heartBeatMap)
110             {
111                 _heartBeatMap[leaseID].stop();
112                 _heartBeatMap.remove(leaseID);
113             }
114             return true;
115         }
116         catch (Throwable e)
117         {
118             logError("RegistryService : ", e.msg);
119             return false;
120         }
121 
122     }
123 
124     Instance[] getAllInstances(string serviceName)
125     {
126         logInfo("getAllInstances : ",serviceName);
127         try
128         {
129             RangeRequest request = new RangeRequest();
130             request.key = cast(ubyte[]) serviceName;
131             // ubyte[] end = cast(ubyte[]) serviceName.dup;
132             // end[$ - 1] += 1;
133             // request.rangeEnd = end;
134             assert(request.key.length != 0);
135 
136             RangeResponse response = _client.Range(request);
137             if (response is null)
138             {
139                 logError("response is null");
140                 return null;
141             }
142             Instance[] values;
143             foreach (v; response.kvs)
144             {
145                 values ~= toObject!Instance(parseJSON(cast(string) v.value));
146             }
147             return values;
148         }
149         catch (Throwable e)
150         {
151             logError("RegistryService : ", e.msg);
152             return null;
153         }
154 
155     }
156 
157     void subscribe(string serviceName, Listener listener)
158     {
159         logInfo("subscribe : ",serviceName);
160         synchronized (this)
161         {
162             _watcher.watch(serviceName, listener);
163         }
164     }
165 
166     void unsubscribe(string serviceName, Listener listener)
167     {
168         logInfo("unsubscribe : ",serviceName);
169         synchronized (this)
170         {
171             _watcher.cancel(serviceName, listener);
172         }
173     }
174 
175 }