Simple UDP layer that just relays messages right now.
| 58 | |
| 59 | |
| 60 | class UDPLayer(layer.Layer): |
| 61 | """ |
| 62 | Simple UDP layer that just relays messages right now. |
| 63 | """ |
| 64 | |
| 65 | flow: udp.UDPFlow | None |
| 66 | |
| 67 | def __init__(self, context: Context, ignore: bool = False): |
| 68 | super().__init__(context) |
| 69 | if ignore: |
| 70 | self.flow = None |
| 71 | else: |
| 72 | self.flow = udp.UDPFlow(self.context.client, self.context.server, True) |
| 73 | |
| 74 | @expect(events.Start) |
| 75 | def start(self, _) -> layer.CommandGenerator[None]: |
| 76 | if self.flow: |
| 77 | yield UdpStartHook(self.flow) |
| 78 | |
| 79 | if self.context.server.timestamp_start is None: |
| 80 | err = yield commands.OpenConnection(self.context.server) |
| 81 | if err: |
| 82 | if self.flow: |
| 83 | self.flow.error = flow.Error(str(err)) |
| 84 | yield UdpErrorHook(self.flow) |
| 85 | yield commands.CloseConnection(self.context.client) |
| 86 | self._handle_event = self.done |
| 87 | return |
| 88 | self._handle_event = self.relay_messages |
| 89 | |
| 90 | _handle_event = start |
| 91 | |
| 92 | @expect(events.DataReceived, events.ConnectionClosed, UdpMessageInjected) |
| 93 | def relay_messages(self, event: events.Event) -> layer.CommandGenerator[None]: |
| 94 | if isinstance(event, UdpMessageInjected): |
| 95 | # we just spoof that we received data here and then process that regularly. |
| 96 | event = events.DataReceived( |
| 97 | self.context.client |
| 98 | if event.message.from_client |
| 99 | else self.context.server, |
| 100 | event.message.content, |
| 101 | ) |
| 102 | |
| 103 | assert isinstance(event, events.ConnectionEvent) |
| 104 | |
| 105 | from_client = event.connection == self.context.client |
| 106 | send_to: Connection |
| 107 | if from_client: |
| 108 | send_to = self.context.server |
| 109 | else: |
| 110 | send_to = self.context.client |
| 111 | |
| 112 | if isinstance(event, events.DataReceived): |
| 113 | if self.flow: |
| 114 | udp_message = udp.UDPMessage(from_client, event.data) |
| 115 | self.flow.messages.append(udp_message) |
| 116 | yield UdpMessageHook(self.flow) |
| 117 | yield commands.SendData(send_to, udp_message.content) |
no outgoing calls
no test coverage detected
searching dependent graphs…