Skip to content

Commit c99d401

Browse files
raphael-goetzraphael-goetz
committed
feat: reworked plugin implementation to match Octopus version 2
1 parent 9c6613c commit c99d401

9 files changed

Lines changed: 359 additions & 151 deletions

File tree

‎README.md‎

Lines changed: 13 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -16,53 +16,24 @@ dependencies {
1616
}
1717
```
1818

19-
Add _depend_ inside `plugin.yml`:
19+
Add _depend on_ inside `plugin.yml`:
2020

2121
```yaml
2222
depend:
2323
- Octopus
2424
```
2525
26-
## Development
26+
### Config
2727
28-
Full development setup available as [Development Container](https://containers.dev/).
29-
Please use it for being able to tell "It works on my machine".
28+
Make sure this is inside of the `/plugins/octopus/config.yml`
3029

31-
**Docker is required to be installed on your machine!**
32-
33-
### Create ~/dev.env
34-
35-
The development container is using a local env file on your
36-
host machine for reading e.g. GitHub Tokens, Usernames, Email.
37-
So please make sure it exists with your credentials in `~/dev.env`:
38-
39-
```text
40-
GITHUB_EMAIL=your-mail@your-domain.com
41-
GITHUB_USERNAME=YOUR_GITHUB_USERNAME
42-
GITHUB_TOKEN=ghp_***
43-
```
44-
45-
The `GITHUB_TOKEN` must've set following permission:
46-
47-
- `repo`
48-
- `read:packages`
49-
- `read:user`
50-
- `user:email`
51-
52-
### IntelliJ IDEA
53-
54-
- Open IntelliJ (Welcome screen)
55-
- Navigate to `Remote Development` - `Dev Containers`
56-
- Press `New Dev Container`
57-
- Select `From VCS Project`
58-
- Select and connect with `Docker`
59-
- Select `IntelliJ IDEA`
60-
- Enter `Git Repository`: `https://github.com/o7studios/octopus-plugin`
61-
- Select `Detection for devcontainer.json file` `Automatic`
62-
- Press `Build Container and Continue`
63-
64-
### Development Container Issues
65-
66-
If you encounter an issue with setting up a development container, please
67-
try to rebuild it first before opening a GitHub Issue. It's not uncommon
68-
that some issues may fix themselves after a fresh container rebuild.
30+
```yml
31+
# Configuration of Octopus-Service
32+
octopus:
33+
# Host of Octopus-gRPC Server
34+
host: "127.0.0.1"
35+
# Port of Octopus-gRPC Server
36+
port: 50051
37+
# Replace to Octopus-API token
38+
token: "development"
39+
```

api/src/main/java/studio/o7/octopus/plugin/api/listener/Listener.java renamed to api/src/main/java/studio/o7/octopus/plugin/api/EventHandler.java

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
1-
package studio.o7.octopus.plugin.api.listener;
1+
package studio.o7.octopus.plugin.api;
22

33
import lombok.AllArgsConstructor;
44
import lombok.Getter;
55
import lombok.NonNull;
6-
import studio.o7.octopus.sdk.gen.api.v1.Object;
6+
import studio.o7.octopus.sdk.v1.Object;
77

88
import java.util.UUID;
99

1010
@AllArgsConstructor
1111
@Getter
12-
public abstract class Listener {
12+
public abstract class EventHandler {
1313
/**
1414
* ID for identifying this listener.
1515
* (Made for internal purposes)
@@ -26,11 +26,6 @@ public abstract class Listener {
2626
*/
2727
protected final String keyPattern;
2828

29-
/**
30-
* Priority of this listener (e.g. determines event order; lower is later)
31-
*/
32-
protected final int priority;
33-
3429
/**
3530
* @param obj The affected object.
3631
*/

‎api/src/main/java/studio/o7/octopus/plugin/api/OctopusError.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package studio.o7.octopus.plugin.api;
2+
3+
import gentle.Error;
4+
import lombok.NonNull;
5+
6+
public enum OctopusError implements Error {
7+
8+
GET_REQUEST_FAILED(0, "While trying to get an entry, a gRPC-Error occurred"),
9+
QUERY_REQUEST_FAILED(1, "While trying to query an entry, a gRPC-Error occurred"),
10+
CALL_REQUEST_FAILED(1, "While trying to call an object, a gRPC-Error occurred"),
11+
12+
;
13+
14+
private final int code;
15+
private final String message;
16+
17+
OctopusError(int code, String message) {
18+
this.code = code;
19+
this.message = message;
20+
}
21+
22+
@Override
23+
public int code() {
24+
return this.code;
25+
}
26+
27+
@Override
28+
public @NonNull String message() {
29+
return this.message;
30+
}
31+
}

‎api/src/main/java/studio/o7/octopus/plugin/api/QueryParameter.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package studio.o7.octopus.plugin.api;
2+
import lombok.Builder;
3+
import lombok.Getter;
4+
5+
@Getter
6+
@Builder
7+
public class QueryParameter {
8+
9+
private String keyPattern;
10+
private String dataFilter;
11+
12+
private boolean includeExpired;
13+
14+
private int page;
15+
private int pageSize;
16+
17+
private com.google.protobuf.Timestamp createdAtStart;
18+
private com.google.protobuf.Timestamp createdAtEnd;
19+
20+
}

‎plugin/src/main/java/studio/o7/octopus/plugin/OctopusImpl.java‎

Lines changed: 93 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -1,132 +1,125 @@
11
package studio.o7.octopus.plugin;
22

3+
import gentle.Error;
4+
import gentle.Result;
35
import io.grpc.stub.StreamObserver;
4-
import it.unimi.dsi.fastutil.Pair;
5-
import it.unimi.dsi.fastutil.objects.Object2ObjectArrayMap;
6-
import it.unimi.dsi.fastutil.objects.Object2ObjectMap;
7-
import lombok.NonNull;
86
import lombok.extern.slf4j.Slf4j;
9-
import org.jetbrains.annotations.NotNull;
10-
import org.jetbrains.annotations.Nullable;
117
import studio.o7.octopus.plugin.api.Octopus;
12-
import studio.o7.octopus.plugin.api.listener.Listener;
13-
import studio.o7.octopus.plugin.observer.EmptyObserver;
14-
import studio.o7.octopus.plugin.utils.ProtoUtils;
15-
import studio.o7.octopus.sdk.OctopusSDK;
16-
import studio.o7.octopus.sdk.gen.api.v1.*;
17-
import studio.o7.octopus.sdk.gen.api.v1.Object;
18-
19-
import java.time.Instant;
20-
import java.util.Collection;
8+
import studio.o7.octopus.plugin.api.EventHandler;
9+
import studio.o7.octopus.plugin.api.OctopusError;
10+
import studio.o7.octopus.plugin.api.QueryParameter;
11+
import studio.o7.octopus.plugin.authentication.OctopusCredentials;
12+
import studio.o7.octopus.plugin.channel.OctopusChannelFactory;
13+
import studio.o7.octopus.plugin.observer.OctopusObserver;
14+
import studio.o7.octopus.sdk.v1.*;
15+
import studio.o7.octopus.sdk.v1.Object;
16+
2117
import java.util.UUID;
22-
import java.util.concurrent.atomic.AtomicReference;
2318

24-
@Slf4j (topic = "OctopusPlugin")
19+
@Slf4j(topic = "OctopusPlugin")
2520
public final class OctopusImpl implements Octopus {
26-
private static final EmptyObserver EMPTY_OBSERVER = new EmptyObserver();
2721

28-
private final OctopusGrpc.OctopusStub stub = OctopusSDK.stub();
29-
private final OctopusGrpc.OctopusBlockingStub blockingStub = OctopusSDK.blockingStub();
30-
private final Object2ObjectMap<UUID, Pair<Listener, StreamObserver<ListenMessage>>> listeners = new Object2ObjectArrayMap<>();
22+
private final OctopusObserver streamObserver;
23+
private StreamObserver<ListenMessage> responseObserver;
24+
25+
private final OctopusGrpc.OctopusBlockingStub blocking;
26+
private final OctopusGrpc.OctopusStub async;
27+
28+
public OctopusImpl(String token, String host, int port) {
29+
var channel = OctopusChannelFactory.getOrCreate(host, port);
30+
31+
var blockingStub = OctopusGrpc.newBlockingStub(channel);
32+
this.blocking = blockingStub.withCallCredentials(new OctopusCredentials(token));
33+
34+
var asyncStub = OctopusGrpc.newStub(channel);
35+
async = asyncStub.withCallCredentials(new OctopusCredentials(token));
36+
37+
this.streamObserver = new OctopusObserver();
38+
}
3139

3240
@Override
33-
public @NotNull Collection<Entry> get(@NonNull String keyPattern, boolean includeExpired, @Nullable Instant createdRangeStart, @Nullable Instant createdRangeEnd) {
34-
var builder = GetRequest.newBuilder();
41+
public Result<Object, Error> get(String key) {
42+
var request = GetRequest.newBuilder().setKey(key).build();
43+
try {
44+
var response = blocking.get(request);
45+
return Result.ok(response.getObject());
46+
} catch (RuntimeException e) {
47+
log.error("Failed to send a get request: {}", e.getMessage());
48+
return Result.err(OctopusError.GET_REQUEST_FAILED);
49+
}
50+
}
3551

36-
builder.setKeyPattern(keyPattern);
37-
builder.setIncludeExpired(includeExpired);
52+
@Override
53+
public Result<QueryResponse, Error> query(QueryParameter queryParameter) {
54+
var request = QueryRequest.newBuilder();
55+
request.setKeyPattern(queryParameter.getKeyPattern());
56+
request.setDataFilter(queryParameter.getDataFilter());
57+
58+
request.setIncludeExpired(queryParameter.isIncludeExpired());
59+
60+
var paginator = Paginator.newBuilder().setPage(queryParameter.getPage()).setPageSize(queryParameter.getPageSize()).build();
61+
62+
request.setPaginator(paginator);
3863

39-
if (createdRangeStart != null)
40-
builder.setCreatedAtRangeStart(ProtoUtils.toProto(createdRangeStart));
64+
if (queryParameter.getCreatedAtStart() != null) {
65+
request.setCreatedAtRangeStart(queryParameter.getCreatedAtStart());
66+
}
4167

42-
if (createdRangeEnd != null)
43-
builder.setCreatedAtRangeEnd(ProtoUtils.toProto(createdRangeEnd));
68+
if (queryParameter.getCreatedAtEnd() != null) {
69+
request.setCreatedAtRangeEnd(queryParameter.getCreatedAtEnd());
70+
}
4471

45-
return this.blockingStub.get(builder.build()).getEntriesList();
72+
try {
73+
var response = blocking.query(request.build());
74+
return Result.ok(response);
75+
} catch (RuntimeException e) {
76+
log.error("Failed to send a query request: {}", e.getMessage());
77+
return Result.err(OctopusError.QUERY_REQUEST_FAILED);
78+
}
4679
}
4780

4881
@Override
49-
public void registerListener(@NonNull Listener listener) {
50-
var requestRef = new AtomicReference<StreamObserver<ListenMessage>>();
51-
52-
var observer = stub.listen(new StreamObserver<>() {
53-
@Override
54-
public void onNext(EventCall value) {
55-
var start = System.currentTimeMillis();
56-
var request = requestRef.get();
57-
if (request == null) return;
58-
59-
listener.onCall(value.getObject());
60-
61-
var msg = ListenMessage.newBuilder()
62-
.setCallback(value)
63-
.build();
64-
65-
request = requestRef.get();
66-
if (request == null) return;
67-
68-
request.onNext(msg);
69-
log.debug("Finished EventCall `{}` in {}ms", value.getCallId(), System.currentTimeMillis() - start);
70-
}
71-
72-
@Override
73-
public void onError(Throwable t) {
74-
requestRef.set(null);
75-
log.error("Cannot call event on listener {} with key-pattern {}", listener.getListenerUniqueId(), listener.getKeyPattern(), t);
76-
unregisterListener(listener);
77-
}
78-
79-
@Override
80-
public void onCompleted() {
81-
requestRef.set(null);
82-
unregisterListener(listener);
83-
log.debug("Completed listener `{}`", listener.getListenerUniqueId());
84-
}
85-
});
86-
87-
requestRef.set(observer);
88-
89-
observer.onNext(ListenMessage.newBuilder()
90-
.setRegister(ListenRegister.newBuilder()
91-
.setKeyPattern(listener.getKeyPattern())
92-
.setPriority(listener.getPriority())
93-
.build()).build());
94-
95-
this.listeners.put(listener.getListenerUniqueId(), new Pair<>() {
96-
@Override
97-
public Listener left() {
98-
return listener;
99-
}
100-
101-
@Override
102-
public StreamObserver<ListenMessage> right() {
103-
return observer;
104-
}
105-
});
82+
public Result<Entry, Error> call(studio.o7.octopus.sdk.v1.Object obj) {
83+
try {
84+
var response = blocking.call(obj);
85+
return Result.ok(response);
86+
} catch (RuntimeException e) {
87+
log.error("Failed to send a call request: {}", e.getMessage());
88+
return Result.err(OctopusError.CALL_REQUEST_FAILED);
89+
}
10690
}
10791

10892
@Override
109-
public void unregisterListener(@NonNull Listener listener) {
110-
unregisterListener(listener.getListenerUniqueId());
93+
public void write(Object obj) {
94+
try {
95+
blocking.write(obj);
96+
} catch (RuntimeException e) {
97+
log.error("Failed to send a write request: {}", e.getMessage());
98+
}
11199
}
112100

113101
@Override
114-
public void unregisterListener(@NonNull UUID listenerUniqueId) {
115-
var pair = this.listeners.get(listenerUniqueId);
116-
if (pair == null) return;
117-
this.listeners.remove(listenerUniqueId);
118-
var right = pair.right();
119-
if (right == null) return;
120-
right.onCompleted();
102+
public void registerHandler(EventHandler eventHandler) {
103+
log.debug("Adding a new handler to the pattern {}", eventHandler.getKeyPattern());
104+
streamObserver.addHandler(eventHandler);
105+
106+
if (responseObserver == null) {
107+
log.debug("initializing stream to octopus");
108+
this.responseObserver = async.listen(streamObserver);
109+
}
110+
111+
var req = ListenMessage.newBuilder().addAllKeyPattern(streamObserver.getKeys());
112+
responseObserver.onNext(req.build());
121113
}
122114

123115
@Override
124-
public @NotNull Entry call(@NonNull Object obj) {
125-
return blockingStub.call(obj);
116+
public void unregisterHandler(@org.jspecify.annotations.NonNull EventHandler eventHandler) {
117+
this.unregisterHandler(eventHandler.getListenerUniqueId());
126118
}
127119

128120
@Override
129-
public void callAndForget(@NonNull Object obj) {
130-
stub.write(obj, EMPTY_OBSERVER);
121+
public void unregisterHandler(UUID listenerUniqueId) {
122+
this.streamObserver.removeHandler(listenerUniqueId);
131123
}
124+
132125
}

0 commit comments

Comments
 (0)