Skip to content

Commit c0f3091

Browse files
mostafamostafa
committed
WIP (#1)
1 parent ba67cf2 commit c0f3091

13 files changed

+510
-327
lines changed

Diff for: ‎Makefile

+2-2
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ run:
22
@poetry run python main.py
33

44
lint:
5-
@buf lint
5+
@poetry run buf lint
66

77
gen:
8-
@buf generate
8+
@poetry run buf generate

Diff for: ‎main.py

+15-158
Original file line numberDiff line numberDiff line change
@@ -1,180 +1,37 @@
1-
import sys
21
import asyncio
3-
import typing
4-
from plugin.v1.plugin_grpc import GatewayDPluginServiceBase
5-
from google.protobuf.struct_pb2 import ListValue, Struct, Value
6-
from google.protobuf.json_format import MessageToDict
7-
import grpclib
8-
from grpclib.server import Server, Stream
9-
from grpclib.reflection.service import ServerReflection
10-
from grpclib.health.service import Health
11-
from grpclib.health.check import ServiceStatus
12-
import logging
13-
14-
logging.basicConfig(filename="plugin.log", level=logging.INFO)
15-
16-
17-
async def defaults(stream: Stream[Struct, Struct]) -> None:
18-
"""Default handler for all hooks.
19-
20-
Note:
21-
The GatewayDPluginServiceBase class implements all hooks as abstract
22-
methods, so we need to implement them all. This is a default handler
23-
to use for all hooks that we don't need to implement.
24-
"""
25-
req = await stream.recv_message()
26-
if req:
27-
await stream.send_message(req)
28-
else:
29-
await stream.send_message(Struct())
30-
31-
32-
class Plugin(GatewayDPluginServiceBase):
33-
async def GetPluginConfig(self, stream: Stream[Struct, Struct]) -> None:
34-
# Ignore the request, as it is empty.
35-
await stream.recv_message()
36-
await stream.send_message(
37-
Struct(
38-
fields={
39-
"id": Value(
40-
struct_value=Struct(
41-
fields={
42-
"name": Value(string_value="plugin-template-python"),
43-
"version": Value(string_value="0.1.0"),
44-
"remoteUrl": Value(
45-
string_value="github.com/gatewayd-io/plugin-template-python"
46-
),
47-
}
48-
)
49-
),
50-
"hooks": Value(
51-
list_value=ListValue(
52-
values=[
53-
# The list of hooks that the plugin implements.
54-
Value(number_value=16), # OnTrafficFromClient
55-
]
56-
)
57-
),
58-
}
59-
)
60-
)
61-
62-
async def OnBooted(self, stream: Stream[Struct, Struct]) -> None:
63-
await defaults(stream)
64-
65-
async def OnBooting(self, stream: Stream[Struct, Struct]) -> None:
66-
await defaults(stream)
67-
68-
async def OnClosed(self, stream: Stream[Struct, Struct]) -> None:
69-
await defaults(stream)
70-
71-
async def OnClosing(self, stream: Stream[Struct, Struct]) -> None:
72-
await defaults(stream)
73-
74-
async def OnConfigLoaded(self, stream: Stream[Struct, Struct]) -> None:
75-
await defaults(stream)
76-
77-
async def OnHook(self, stream: Stream[Struct, Struct]) -> None:
78-
await defaults(stream)
79-
80-
async def OnNewClient(self, stream: Stream[Struct, Struct]) -> None:
81-
await defaults(stream)
82-
83-
async def OnNewLogger(self, stream: Stream[Struct, Struct]) -> None:
84-
await defaults(stream)
85-
86-
async def OnNewPool(self, stream: Stream[Struct, Struct]) -> None:
87-
await defaults(stream)
88-
89-
async def OnNewProxy(self, stream: Stream[Struct, Struct]) -> None:
90-
await defaults(stream)
91-
92-
async def OnNewServer(self, stream: Stream[Struct, Struct]) -> None:
93-
await defaults(stream)
94-
95-
async def OnOpened(self, stream: Stream[Struct, Struct]) -> None:
96-
await defaults(stream)
97-
98-
async def OnOpening(self, stream: Stream[Struct, Struct]) -> None:
99-
await defaults(stream)
100-
101-
async def OnRun(self, stream: Stream[Struct, Struct]) -> None:
102-
await defaults(stream)
103-
104-
async def OnShutdown(self, stream: Stream[Struct, Struct]) -> None:
105-
await defaults(stream)
106-
107-
async def OnSignal(self, stream: Stream[Struct, Struct]) -> None:
108-
await defaults(stream)
109-
110-
async def OnTick(self, stream: Stream[Struct, Struct]) -> None:
111-
await defaults(stream)
112-
113-
async def OnTraffic(self, stream: Stream[Struct, Struct]) -> None:
114-
await defaults(stream)
115-
116-
async def OnTrafficFromClient(self, stream: Stream[Struct, Struct]) -> None:
117-
"""
118-
This is an example of how to use the OnTrafficFromClient hook to
119-
intercept traffic from the client and modify it before it is sent to
120-
the server. In this example, we simply log the request and send it
121-
to the server.
122-
"""
123-
req = await stream.recv_message()
124-
if req:
125-
logging.info(MessageToDict(req))
126-
await stream.send_message(req)
127-
else:
128-
await stream.send_message(Struct())
129-
130-
async def OnTrafficFromServer(self, stream: Stream[Struct, Struct]) -> None:
131-
await defaults(stream)
132-
133-
async def OnTrafficToClient(self, stream: Stream[Struct, Struct]) -> None:
134-
await defaults(stream)
135-
136-
async def OnTrafficToServer(self, stream: Stream[Struct, Struct]) -> None:
137-
await defaults(stream)
138-
139-
def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
140-
"""Mapping of service methods to handlers.
141-
142-
Note:
143-
This method is used to remap the service name to the plugin name,
144-
so that the plugin health check works properly.
145-
"""
146-
# Get the mapping from the base class
147-
mapping = super().__mapping__()
148-
remapping = {}
149-
150-
# Replace the service name with the plugin name
151-
for path in mapping:
152-
repath = path.replace("plugin.v1.GatewayDPluginService", "plugin")
153-
remapping[repath] = mapping[path]
2+
import sys
1543

155-
# Merge the two mappings
156-
remapping.update(mapping)
4+
import structlog
5+
from grpclib.health.check import ServiceStatus
6+
from grpclib.health.service import Health
7+
from grpclib.reflection.service import ServerReflection
8+
from grpclib.server import Server
9+
from module import Plugin
10+
from stdio_service import GRPCStdioService
15711

158-
return remapping
12+
logger = structlog.get_logger()
15913

16014

16115
async def serve() -> None:
16216
# Instantiate the plugin.
16317
plugin = Plugin()
16418

19+
# Instantiate the stdio service.
20+
stdio = GRPCStdioService()
21+
16522
# Create a health check for the plugin.
16623
plugin_health = ServiceStatus()
16724
plugin_health.set(True)
16825
health = Health({plugin: [plugin_health]})
16926

17027
# Add reflection for the plugin and health check.
171-
services = ServerReflection.extend([plugin, health])
28+
services = ServerReflection.extend([plugin, stdio, health])
17229

17330
# Instantiate the server.
17431
server = Server(services)
17532

17633
# Start the server.
177-
await server.start("127.0.0.1", 12345)
34+
await server.start("127.0.0.1", 50001)
17835
await server.wait_closed()
17936

18037

@@ -183,7 +40,7 @@ async def serve() -> None:
18340
asyncio.set_event_loop(loop)
18441
try:
18542
# This is a special message that tells gatewayd to start the plugin.
186-
print("1|0|tcp|127.0.0.1:12345|grpc")
43+
print("1|0|tcp|127.0.0.1:50001|grpc")
18744
sys.stdout.flush()
18845
asyncio.run(serve())
18946
except KeyboardInterrupt:

Diff for: ‎mapping.py

+25
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
import typing
2+
import grpclib
3+
4+
5+
class MappingMixin:
6+
def __mapping__(self) -> typing.Dict[str, grpclib.const.Handler]:
7+
"""Mapping of service methods to handlers.
8+
9+
Note:
10+
This method is used to remap the service name to the plugin name,
11+
so that the plugin health check works properly.
12+
"""
13+
# Get the mapping from the base class
14+
mapping = super().__mapping__() # type: ignore
15+
remapping = {}
16+
17+
# Replace the service name with the plugin name
18+
for path in mapping:
19+
repath = path.replace("plugin.v1.GatewayDPluginService", "plugin")
20+
remapping[repath] = mapping[path]
21+
22+
# Merge the two mappings
23+
remapping.update(mapping)
24+
25+
return remapping

Diff for: ‎module.py

+132
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,132 @@
1+
import structlog
2+
from google.protobuf.json_format import MessageToDict
3+
from google.protobuf.struct_pb2 import ListValue, Struct, Value
4+
from grpclib.server import Stream
5+
6+
from plugin.v1.plugin_grpc import GatewayDPluginServiceBase
7+
from mapping import MappingMixin
8+
9+
logger = structlog.get_logger()
10+
11+
12+
async def defaults(stream: Stream[Struct, Struct]) -> None:
13+
"""Default handler for all hooks.
14+
15+
Note:
16+
The GatewayDPluginServiceBase class implements all hooks as abstract
17+
methods, so we need to implement them all. This is a default handler
18+
to use for all hooks that we don't need to implement.
19+
"""
20+
req = await stream.recv_message()
21+
if req:
22+
await stream.send_message(req)
23+
else:
24+
await stream.send_message(Struct())
25+
26+
27+
class Plugin(MappingMixin, GatewayDPluginServiceBase):
28+
async def GetPluginConfig(self, stream: Stream[Struct, Struct]) -> None:
29+
# Ignore the request, as it is empty.
30+
await stream.recv_message()
31+
await stream.send_message(
32+
Struct(
33+
fields={
34+
"id": Value(
35+
struct_value=Struct(
36+
fields={
37+
"name": Value(string_value="plugin-template-python"),
38+
"version": Value(string_value="0.1.0"),
39+
"remoteUrl": Value(
40+
string_value="github.com/gatewayd-io/plugin-template-python"
41+
),
42+
}
43+
)
44+
),
45+
"hooks": Value(
46+
list_value=ListValue(
47+
values=[
48+
# The list of hooks that the plugin implements.
49+
Value(number_value=16), # OnTrafficFromClient
50+
]
51+
)
52+
),
53+
}
54+
)
55+
)
56+
57+
async def OnBooted(self, stream: Stream[Struct, Struct]) -> None:
58+
await defaults(stream)
59+
60+
async def OnBooting(self, stream: Stream[Struct, Struct]) -> None:
61+
await defaults(stream)
62+
63+
async def OnClosed(self, stream: Stream[Struct, Struct]) -> None:
64+
await defaults(stream)
65+
66+
async def OnClosing(self, stream: Stream[Struct, Struct]) -> None:
67+
await defaults(stream)
68+
69+
async def OnConfigLoaded(self, stream: Stream[Struct, Struct]) -> None:
70+
await defaults(stream)
71+
72+
async def OnHook(self, stream: Stream[Struct, Struct]) -> None:
73+
await defaults(stream)
74+
75+
async def OnNewClient(self, stream: Stream[Struct, Struct]) -> None:
76+
await defaults(stream)
77+
78+
async def OnNewLogger(self, stream: Stream[Struct, Struct]) -> None:
79+
await defaults(stream)
80+
81+
async def OnNewPool(self, stream: Stream[Struct, Struct]) -> None:
82+
await defaults(stream)
83+
84+
async def OnNewProxy(self, stream: Stream[Struct, Struct]) -> None:
85+
await defaults(stream)
86+
87+
async def OnNewServer(self, stream: Stream[Struct, Struct]) -> None:
88+
await defaults(stream)
89+
90+
async def OnOpened(self, stream: Stream[Struct, Struct]) -> None:
91+
await defaults(stream)
92+
93+
async def OnOpening(self, stream: Stream[Struct, Struct]) -> None:
94+
await defaults(stream)
95+
96+
async def OnRun(self, stream: Stream[Struct, Struct]) -> None:
97+
await defaults(stream)
98+
99+
async def OnShutdown(self, stream: Stream[Struct, Struct]) -> None:
100+
await defaults(stream)
101+
102+
async def OnSignal(self, stream: Stream[Struct, Struct]) -> None:
103+
await defaults(stream)
104+
105+
async def OnTick(self, stream: Stream[Struct, Struct]) -> None:
106+
await defaults(stream)
107+
108+
async def OnTraffic(self, stream: Stream[Struct, Struct]) -> None:
109+
await defaults(stream)
110+
111+
async def OnTrafficFromClient(self, stream: Stream[Struct, Struct]) -> None:
112+
"""
113+
This is an example of how to use the OnTrafficFromClient hook to
114+
intercept traffic from the client and modify it before it is sent to
115+
the server. In this example, we simply log the request and send it
116+
to the server.
117+
"""
118+
req = await stream.recv_message()
119+
if req:
120+
logger.info(MessageToDict(req))
121+
await stream.send_message(req)
122+
else:
123+
await stream.send_message(Struct())
124+
125+
async def OnTrafficFromServer(self, stream: Stream[Struct, Struct]) -> None:
126+
await defaults(stream)
127+
128+
async def OnTrafficToClient(self, stream: Stream[Struct, Struct]) -> None:
129+
await defaults(stream)
130+
131+
async def OnTrafficToServer(self, stream: Stream[Struct, Struct]) -> None:
132+
await defaults(stream)

0 commit comments

Comments
 (0)