neton.client.Watcher source code

1 module neton.client.Watcher;
2 
3 import neton.protocol.neton;
4 import neton.protocol.neton;
5 import neton.protocol.netonrpc;
6 import neton.client.Listener;
7 import neton.client.Event;
8 import neton.util.future;
9 import grpc;
10 import hunt.logging;
11 import core.thread;
12 import std.algorithm.mutation;
13 import std.stdio;
14 
15 class Watcher
16 {
17     private
18     {
19         string _prefix;
20 
21         ClientReaderWriter!(WatchResponse, WatchRequest) _stream;
22         Future!(int, WatchResponse) _createWatch;
23         Future!(int, WatchResponse) _cancelWatch;
24         Listener[][long] _listenMap;
25         long[string] _keyWatchIdMap;
26 
27     }
28 
29     long watch(string key, Listener lit)
30     {
31         writefln("watch key: %s , listener :%s ", key, lit);
32         string configKey = _prefix ~ "/" ~ key;
33         if (key in _keyWatchIdMap)
34         {
35             auto id = _keyWatchIdMap[key];
36             if (id in _listenMap)
37             {
38                 _listenMap[id] ~= lit;
39                 return id;
40             }
41             else
42             {
43                 _keyWatchIdMap.remove(key);
44             }
45         }
46 
47         WatchRequest request = new WatchRequest();
48         request._requestUnionCase = WatchRequest.RequestUnionCase.createRequest;
49         request._createRequest = new WatchCreateRequest();
50         request._createRequest.key = cast(ubyte[]) configKey;
51         ubyte[] end = cast(ubyte[]) configKey.dup;
52         request._createRequest.rangeEnd = end;
53         end[$ - 1] += 1;
54         _stream.write(request);
55 
56         auto f = new Future!(int, WatchResponse)(0);
57         _createWatch = f;
58         WatchResponse response = f.get();
59         auto id = response.watchId;
60         _listenMap[id] ~= lit;
61         _keyWatchIdMap[key] = id;
62         return id;
63     }
64 
65     bool cancel(string key, Listener listener)
66     {
67         writefln("cancel watch key: %s , listener :%s ", key, listener);
68 
69         long id;
70         if (key in _keyWatchIdMap)
71         {
72             id = _keyWatchIdMap[key];
73             // logError("before cancel size : ", _listenMap[id].length);
74             _listenMap[id] = remove!(a => a is listener)(_listenMap[id]);
75             // logError("after cancel size : ", _listenMap[id].length);
76             if (_listenMap[id].length > 0)
77                 return true;
78         }
79         else
80             return false;
81         WatchRequest request = new WatchRequest();
82         request._requestUnionCase = WatchRequest.RequestUnionCase.cancelRequest;
83         request._cancelRequest = new WatchCancelRequest();
84         request._cancelRequest.watchId = id;
85         _stream.write(request);
86         auto f = new Future!(int, WatchResponse)(0);
87         _cancelWatch = f;
88         f.get();
89         return true;
90     }
91 
92     this(ClientReaderWriter!(WatchResponse, WatchRequest) stream, string prefix)
93     {
94         this._stream = stream;
95         this._prefix = prefix;
96 
97         new Thread(() {
98             WatchResponse response;
99             try
100             {
101                 while (_stream.read(response))
102                 {
103                     // logError("begin ",);
104                     if (response.created)
105                     {
106                         // logError("created ", response.watchId);
107                         _createWatch.done(response);
108                     }
109                     else if (response.canceled)
110                     {
111                         // logError("canceled ", response.watchId);
112                         _cancelWatch.done(response);
113                     }
114                     else
115                     {
116                         // logError("other ");
117                         foreach (neton.protocol.neton.Event e; response.events)
118                         {
119                             neton.client.Event.Event event;
120                             event.key = cast(string) e.kv.key;
121                             event.value = cast(string) e.kv.value;
122                             auto type = cast(int)(e.type);
123                             event.type = cast(neton.client.Event.Event.EventType) type;
124                             auto id = response.watchId;
125                             // logError("errr start ", response.watchId);
126                             if (id in _listenMap)
127                             {
128                                 foreach (listener; _listenMap[id])
129                                     listener.onEvent(event);
130                             }
131                             // logError("errr end ", response.watchId);
132                         }
133                     }
134                     // logError("end");
135                 }
136             }
137             catch (Throwable e)
138             {
139                 logError(e.msg);
140             }
141             // logError("in false");
142         }).start();
143     }
144 
145     bool close()
146     {
147         _stream.writesDone();
148         return _stream.finish().ok();
149     }
150 
151 }