• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

leeonky / test-charm-java / 144

05 Mar 2025 01:08PM UTC coverage: 74.367% (+0.07%) from 74.297%
144

push

circleci

leeonky
Update version

7932 of 10666 relevant lines covered (74.37%)

0.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.48
/DAL-extension-inspector/src/main/java/com/github/leeonky/dal/extensions/inspector/Inspector.java
1
package com.github.leeonky.dal.extensions.inspector;
2

3
import com.github.leeonky.dal.DAL;
4
import com.github.leeonky.dal.runtime.Data;
5
import com.github.leeonky.dal.runtime.RuntimeContextBuilder.DALRuntimeContext;
6
import com.github.leeonky.interpreter.InterpreterException;
7
import com.github.leeonky.util.Suppressor;
8
import io.javalin.Javalin;
9
import io.javalin.http.Context;
10
import io.javalin.http.staticfiles.Location;
11
import io.javalin.websocket.WsContext;
12

13
import java.io.ByteArrayInputStream;
14
import java.io.ByteArrayOutputStream;
15
import java.io.InputStream;
16
import java.net.InetAddress;
17
import java.net.NetworkInterface;
18
import java.time.Instant;
19
import java.util.*;
20
import java.util.concurrent.ConcurrentHashMap;
21
import java.util.concurrent.CountDownLatch;
22
import java.util.function.Supplier;
23
import java.util.stream.Collectors;
24

25
import static com.github.leeonky.util.function.Extension.getFirstPresent;
26
import static java.net.URLConnection.guessContentTypeFromStream;
27
import static java.util.Objects.requireNonNull;
28
import static java.util.Optional.ofNullable;
29

30
public class Inspector {
31
    private static Inspector inspector = null;
1✔
32
    private static Mode mode = null;
1✔
33
    private final Javalin javalin;
34
    private final CountDownLatch serverReadyLatch = new CountDownLatch(1);
1✔
35
    private final Set<DAL> instances = new LinkedHashSet<>();
1✔
36
    //   TODO refactor
37
    private final Map<String, WsContext> clientConnections = new ConcurrentHashMap<>();
1✔
38
    private final Map<String, Set<String>> clientMonitors = new ConcurrentHashMap<>();
1✔
39
    private final Map<String, DalInstance> dalInstances = new ConcurrentHashMap<>();
1✔
40
    private static Supplier<Object> defaultInput = () -> null;
1✔
41

42
    public Inspector() {
1✔
43
        DalInstance defaultInstance = new DalInstance(() -> defaultInput.get(), DAL.create("Try It!", InspectorExtension.class), "");
1✔
44
        defaultInstance.running = false;
1✔
45
        dalInstances.put("Try It!", defaultInstance);
1✔
46
        javalin = Javalin.create(config -> config.addStaticFiles("/public", Location.CLASSPATH))
1✔
47
                .events(event -> event.serverStarted(serverReadyLatch::countDown));
1✔
48
        requireNonNull(javalin.jettyServer()).setServerPort(getServerPort());
1✔
49
        javalin.get("/", ctx -> ctx.redirect("/index.html"));
1✔
50
        javalin.post("/api/execute", ctx -> ctx.html(execute(ctx.queryParam("name"), ctx.body())));
1✔
51
        javalin.post("/api/exchange", ctx -> exchange(ctx.queryParam("session"), ctx.body()));
1✔
52
        javalin.post("/api/pass", ctx -> pass(ctx.queryParam("name")));
1✔
53
        javalin.post("/api/release", ctx -> release(ctx.queryParam("name")));
1✔
54
        javalin.post("/api/release-all", ctx -> releaseAll());
1✔
55
        javalin.get("/api/request", ctx -> ctx.html(request(ctx.queryParam("name"))));
1✔
56
        javalin.get("/attachments", ctx -> responseAttachment(ctx.queryParam("name"), Integer.parseInt(ctx.queryParam("index")), ctx));
1✔
57
        javalin.ws("/ws/exchange", ws -> {
1✔
58
            ws.onConnect(ctx -> {
1✔
59
                clientConnections.put(ctx.getSessionId(), ctx);
1✔
60
                sendInstances(ctx);
1✔
61
            });
1✔
62
            ws.onClose(ctx -> clientConnections.remove(ctx.getSessionId()));
1✔
63
        });
1✔
64
        javalin.start();
1✔
65
    }
1✔
66

67
    private void responseAttachment(String name, int index, Context ctx) {
68
        DalInstance dalInstance = dalInstances.get(name);
1✔
69
        if (dalInstance != null) {
1✔
70
            Watch watch = dalInstance.watches.get(index);
1✔
71
            if (watch instanceof DalInstance.BinaryWatch) {
1✔
72
                DalInstance.BinaryWatch binaryWatch = (DalInstance.BinaryWatch) watch;
1✔
73
                String contentType = Suppressor.get(() -> guessContentTypeFromStream(
1✔
74
                        new ByteArrayInputStream(binaryWatch.binary())));
1✔
75
                ctx.contentType(contentType == null ? "application/octet-stream" : contentType);
1✔
76
                ctx.result(binaryWatch.binary());
1✔
77
            }
78
        }
79
    }
1✔
80

81
    public static void watch(DAL dal, String property, Data value) {
82
        if (inspector != null)
1✔
83
            inspector.watchInner(dal, property, value);
1✔
84
    }
1✔
85

86
    private void watchInner(DAL dal, String property, Data value) {
87
        if (inspector.calledFromInspector())
1✔
88
            dalInstances.get(dal.getName()).watch(property, value);
1✔
89
    }
1✔
90

91
    private void pass(String name) {
92
        if (!name.equals("Try It!")) {
1✔
93
            DalInstance remove = dalInstances.remove(name);
1✔
94
            if (remove != null)
1✔
95
                remove.pass();
1✔
96
        }
97
    }
1✔
98

99
    private void waitForReady() {
100
        Suppressor.run(serverReadyLatch::await);
1✔
101
    }
1✔
102

103
    private static int getServerPort() {
104

105
        return getFirstPresent(() -> ofNullable(System.getenv("DAL_INSPECTOR_PORT")),
1✔
106
                () -> ofNullable(System.getProperty("dal.inspector.port")))
1✔
107
                .map(Integer::parseInt)
1✔
108
                .orElse(10082);
1✔
109
    }
110

111
    public static void ready() {
112
        inspector.waitForReady();
1✔
113
    }
1✔
114

115
    private void releaseAll() {
116
        for (String instanceName : new ArrayList<>(dalInstances.keySet()))
1✔
117
            release(instanceName);
1✔
118
    }
1✔
119

120
    private void release(String name) {
121
        if (!name.equals("Try It!")) {
1✔
122
            DalInstance remove = dalInstances.remove(name);
1✔
123
            if (remove != null)
1✔
124
                remove.release();
1✔
125
        }
126
    }
1✔
127

128
    public static void setDefaultMode(Mode mode) {
129
        Inspector.mode = mode;
1✔
130
    }
1✔
131

132
    private void exchange(String session, String body) {
133
        if (clientConnections.containsKey(session)) {
1✔
134
            clientMonitors.put(session, Arrays.stream(body.trim().split("\\n")).map(String::trim).filter(s -> !s.isEmpty()).collect(Collectors.toSet()));
1✔
135

136
            for (DalInstance dalInstance : dalInstances.values()) {
1✔
137
                if (dalInstance.running)
1✔
138
                    clientConnections.get(session).send(ObjectWriter.serialize(new HashMap<String, String>() {{
1✔
139
                        put("request", dalInstance.dal.getName());
1✔
140
                    }}));
1✔
141
            }
1✔
142
        }
143
    }
1✔
144

145
    public static class DalInstance {
146
        private final Supplier<Object> input;
147
        private boolean running = true;
1✔
148
        private boolean pass = false;
1✔
149
        private final DAL dal;
150
        private final String code;
151
        private final List<Watch> watches = new ArrayList<>();
1✔
152

153
        public DalInstance(Supplier<Object> input, DAL dal, String code) {
1✔
154
            this.input = input;
1✔
155
            this.dal = dal;
1✔
156
            this.code = code;
1✔
157
        }
1✔
158

159
        public String execute(String code) {
160
            watches.clear();
1✔
161
            Map<String, Object> response = new HashMap<>();
1✔
162
            Object inputObject = input.get();
1✔
163
            DALRuntimeContext runtimeContext = dal.getRuntimeContextBuilder().build(inputObject);
1✔
164
            try {
165
                response.put("root", runtimeContext.wrap(inputObject).dumpAll());
1✔
166
                response.put("inspect", dal.compileSingle(code, runtimeContext).inspect());
1✔
167
                response.put("result", runtimeContext.wrap(dal.evaluate(inputObject, code)).dumpAll());
1✔
168
            } catch (InterpreterException e) {
1✔
169
                response.put("error", e.show(code) + "\n\n" + e.getMessage());
1✔
170
            }
1✔
171
            response.put("watches", watches.stream().map(Watch::collect).collect(Collectors.toList()));
1✔
172
            return ObjectWriter.serialize(response);
1✔
173
        }
174

175
        public boolean hold() {
176
            System.err.println("Waiting for DAL inspector release...");
1✔
177
            try {
178
                Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
1✔
179

180
                System.err.println("\tDal inspector running at:");
1✔
181

182
                while (interfaces.hasMoreElements()) {
1✔
183
                    Enumeration<InetAddress> inetAddresses = interfaces.nextElement().getInetAddresses();
1✔
184
                    while (inetAddresses.hasMoreElements()) {
1✔
185
                        InetAddress address = inetAddresses.nextElement();
1✔
186
                        System.err.printf("\t\thttp://%s:%d%n", address.getHostAddress(), getServerPort());
1✔
187
                    }
1✔
188
                }
1✔
189
            } catch (Exception ignore) {
×
190
            }
1✔
191
            //        TODO use sempahore to wait for the result
192
            while (running)
1✔
193
                Suppressor.run(() -> Thread.sleep(20));
1✔
194
//            TODO use logger
195
            System.err.println("DAL inspector released with pass: " + pass);
1✔
196
            return pass;
1✔
197
        }
198

199
        public void release() {
200
            running = false;
1✔
201
        }
1✔
202

203
        public void pass() {
204
            pass = true;
1✔
205
            release();
1✔
206
        }
1✔
207

208
        private byte[] getBytes(Data data) {
209
            if (data.instance() instanceof byte[])
1✔
210
                return (byte[]) data.instance();
1✔
211
            if (data.instance() instanceof InputStream) {
1✔
212
                InputStream stream = (InputStream) data.instance();
×
213
                return Suppressor.get(() -> {
×
214
                    try (ByteArrayOutputStream buffer = new ByteArrayOutputStream()) {
×
215
                        int size;
216
                        byte[] data1 = new byte[1024];
×
217
                        while ((size = stream.read(data1, 0, data1.length)) != -1)
×
218
                            buffer.write(data1, 0, size);
×
219
                        return buffer.toByteArray();
×
220
                    }
221
                });
222
            }
223
            if (data.instance() instanceof Byte[]) {
1✔
224
                ByteArrayOutputStream stream = new ByteArrayOutputStream();
×
225
                data.list().values().forEach(b -> stream.write((byte) b));
×
226
                return stream.toByteArray();
×
227
            }
228
            return null;
1✔
229
        }
230

231
        public void watch(String property, Data value) {
232
            byte[] bytes = getBytes(value);
1✔
233
            if (bytes != null) {
1✔
234
                watches.add(new BinaryWatch(property, bytes));
1✔
235
            } else
236
                watches.add(new DefaultWatch(property, value));
1✔
237
        }
1✔
238

239

240
        public class DefaultWatch implements Watch {
241
            private final String property;
242
            private final String value;
243

244
            public DefaultWatch(String property, Data value) {
1✔
245
                this.property = property;
1✔
246
                this.value = value.dumpAll();
1✔
247
            }
1✔
248

249
            @Override
250
            public Map<String, Object> collect() {
251
                return new HashMap<String, Object>() {{
1✔
252
                    put("property", property);
1✔
253
                    put("type", "DEFAULT");
1✔
254
                    put("value", value);
1✔
255
                }};
1✔
256
            }
257
        }
258

259
        private class BinaryWatch implements Watch {
260
            private final String property;
261
            private final int index;
262
            private final byte[] binary;
263

264
            public BinaryWatch(String property, byte[] value) {
1✔
265
                this.property = property;
1✔
266
                index = watches.size();
1✔
267
                binary = new byte[value.length];
1✔
268
                System.arraycopy(value, 0, binary, 0, value.length);
1✔
269
            }
1✔
270

271
            public byte[] binary() {
272
                return binary;
1✔
273
            }
274

275
            @Override
276
            public Map<String, Object> collect() {
277
                return new HashMap<String, Object>() {{
1✔
278
                    put("property", property);
1✔
279
                    put("type", "BINARY");
1✔
280
                    put("url", "/attachments?name=" + dal.getName() + "&index=" + index + "&tm=" + Instant.now().getEpochSecond());
1✔
281
                }};
1✔
282
            }
283
        }
284
    }
285

286
    public boolean inspectInner(DAL dal, Object input, String code) {
287
        if (calledFromInspector())
1✔
288
            return false;
1✔
289
//        lock inspect by name
290
//        check mode
291
        if (currentMode() == Mode.FORCED) {
1✔
292
            DalInstance dalInstance = new DalInstance(() -> input, dal, code);
1✔
293
            dalInstances.put(dal.getName(), dalInstance);
1✔
294

295
//            List<WsContext> monitored = clientMonitors.entrySet().stream().filter(e -> e.getValue().contains(dal.getName()))
296
//                    .map(o -> clientConnections.get(o.getKey()))
297
//                    .collect(Collectors.toList());
298
//            TODO check monitor flag
299
            for (WsContext wsContext : clientConnections.values()) {
1✔
300
                wsContext.send(ObjectWriter.serialize(new HashMap<String, String>() {{
1✔
301
                    put("request", dal.getName());
1✔
302
                }}));
1✔
303
            }
1✔
304

305
            return dalInstance.hold();
1✔
306

307
        } else {
308
//        TODO refactor
309
            List<WsContext> monitored = clientMonitors.entrySet().stream().filter(e -> e.getValue().contains(dal.getName()))
1✔
310
                    .map(o -> clientConnections.get(o.getKey()))
1✔
311
                    .collect(Collectors.toList());
1✔
312
            if (!monitored.isEmpty()) {
1✔
313
                DalInstance dalInstance = new DalInstance(() -> input, dal, code);
1✔
314
                dalInstances.put(dal.getName(), dalInstance);
1✔
315
                for (WsContext wsContext : monitored) {
1✔
316
                    wsContext.send(ObjectWriter.serialize(new HashMap<String, String>() {{
1✔
317
                        put("request", dal.getName());
1✔
318
                    }}));
1✔
319
                }
1✔
320
                return dalInstance.hold();
×
321
            }
322
            return false;
1✔
323
        }
324
    }
325

326
    public static boolean inspect(DAL dal, Object input, String code) {
327
        if (currentMode() != Mode.DISABLED)
1✔
328
            return inspector.inspectInner(dal, input, code);
1✔
329
        return false;
×
330
    }
331

332
    private String request(String name) {
333
//       TODO reject other request
334
        return dalInstances.get(name).code;
1✔
335
    }
336

337
    private String execute(String name, String code) {
338
        return dalInstances.get(name).execute(code);
1✔
339
    }
340

341
    public static void register(DAL dal) {
342
        inspector.addInstance(dal);
1✔
343
    }
1✔
344

345
    private void addInstance(DAL dal) {
346
        instances.add(dal);
1✔
347
        for (WsContext ctx : clientConnections.values()) {
1✔
348
            sendInstances(ctx);
1✔
349
        }
1✔
350
    }
1✔
351

352
    private void sendInstances(WsContext ctx) {
353
        ctx.send(ObjectWriter.serialize(new HashMap<String, Object>() {{
1✔
354
            put("instances", instances.stream().map(DAL::getName).collect(Collectors.toSet()));
1✔
355
            put("session", ctx.getSessionId());
1✔
356
        }}));
1✔
357
    }
1✔
358

359
    private void stop() {
360
        javalin.close();
1✔
361
    }
1✔
362

363
    public static void launch() {
364
        if (inspector == null) {
1✔
365
            inspector = new Inspector();
1✔
366
        }
367
    }
1✔
368

369
    public static void shutdown() {
370
        if (inspector != null) {
1✔
371
            inspector.stop();
1✔
372
            inspector = null;
1✔
373
        }
374
    }
1✔
375

376
    public static void setDefaultInput(Supplier<Object> supplier) {
377
        defaultInput = supplier;
1✔
378
    }
1✔
379

380
    public static Mode currentMode() {
381
        return getFirstPresent(() -> ofNullable(mode),
1✔
382
                () -> ofNullable(System.getenv("DAL_INSPECTOR_MODE")).map(Mode::valueOf),
×
383
                () -> ofNullable(System.getProperty("dal.inspector.mode")).map(Mode::valueOf))
×
384
                .orElse(Mode.DISABLED);
1✔
385
    }
386

387
    public enum Mode {
1✔
388
        DISABLED, FORCED, AUTO
1✔
389
    }
390

391
    private boolean calledFromInspector() {
392
        for (StackTraceElement stack : Thread.currentThread().getStackTrace())
1✔
393
            if (DalInstance.class.getName().equals(stack.getClassName()))
1✔
394
                return true;
1✔
395
        return false;
1✔
396
    }
397

398
    public static void main(String[] args) {
399
        launch();
×
400
    }
×
401

402
    interface Watch {
403
        Map<String, Object> collect();
404
    }
405
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc