Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 1 | # |
| 2 | # Copyright 2018 the original author or authors. |
| 3 | # |
| 4 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | # you may not use this file except in compliance with the License. |
| 6 | # You may obtain a copy of the License at |
| 7 | # |
| 8 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | # |
| 10 | # Unless required by applicable law or agreed to in writing, software |
| 11 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | # See the License for the specific language governing permissions and |
| 14 | # limitations under the License. |
| 15 | # |
| 16 | |
| 17 | """ |
| 18 | Agent to play gateway between adapters. |
| 19 | """ |
| 20 | |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 21 | from __future__ import absolute_import |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 22 | import structlog |
| 23 | from uuid import uuid4 |
| 24 | from twisted.internet.defer import inlineCallbacks, returnValue |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 25 | from .container_proxy import ContainerProxy |
William Kurkian | ede82e9 | 2019-03-05 13:02:57 -0500 | [diff] [blame] | 26 | from voltha_protos.inter_container_pb2 import InterAdapterHeader, \ |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 27 | InterAdapterMessage |
| 28 | import time |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 29 | import codecs |
| 30 | import six |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 31 | |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 32 | log = structlog.get_logger() |
| 33 | |
| 34 | |
| 35 | class AdapterProxy(ContainerProxy): |
| 36 | |
serkant.uluderya | c344f30 | 2020-05-29 23:26:57 -0700 | [diff] [blame^] | 37 | def __init__(self, kafka_proxy, adapter_topic, my_listening_topic): |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 38 | super(AdapterProxy, self).__init__(kafka_proxy, |
serkant.uluderya | c344f30 | 2020-05-29 23:26:57 -0700 | [diff] [blame^] | 39 | adapter_topic, |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 40 | my_listening_topic) |
| 41 | |
| 42 | def _to_string(self, unicode_str): |
| 43 | if unicode_str is not None: |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 44 | if isinstance(unicode_str, six.string_types): |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 45 | return unicode_str |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 46 | else: |
| 47 | return codecs.encode(unicode_str, 'ascii') |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 48 | else: |
Zack Williams | 84a71e9 | 2019-11-15 09:00:19 -0700 | [diff] [blame] | 49 | return None |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 50 | |
| 51 | @ContainerProxy.wrap_request(None) |
| 52 | @inlineCallbacks |
| 53 | def send_inter_adapter_message(self, |
| 54 | msg, |
| 55 | type, |
| 56 | from_adapter, |
| 57 | to_adapter, |
| 58 | to_device_id=None, |
| 59 | proxy_device_id=None, |
| 60 | message_id=None): |
| 61 | """ |
| 62 | Sends a message directly to an adapter. This is typically used to send |
| 63 | proxied messages from one adapter to another. An initial ACK response |
| 64 | is sent back to the invoking adapter. If there is subsequent response |
| 65 | to be sent back (async) then the adapter receiving this request will |
| 66 | use this same API to send back the async response. |
| 67 | :param msg : GRPC message to send |
| 68 | :param type : InterAdapterMessageType of the message to send |
| 69 | :param from_adapter: Name of the adapter making the request. |
| 70 | :param to_adapter: Name of the remote adapter. |
| 71 | :param to_device_id: The ID of the device for to the message is |
| 72 | intended. if it's None then the message is not intended to a specific |
| 73 | device. Its interpretation is adapter specific. |
| 74 | :param proxy_device_id: The ID of the device which will proxy that |
| 75 | message. If it's None then there is no specific device to proxy the |
| 76 | message. Its interpretation is adapter specific. |
| 77 | :param message_id: A unique number for this transaction that the |
| 78 | adapter may use to correlate a request and an async response. |
| 79 | """ |
| 80 | |
| 81 | try: |
| 82 | # validate params |
| 83 | assert msg |
| 84 | assert from_adapter |
| 85 | assert to_adapter |
| 86 | |
| 87 | # Build the inter adapter message |
| 88 | h = InterAdapterHeader() |
| 89 | h.type = type |
| 90 | h.from_topic = self._to_string(from_adapter) |
serkant.uluderya | c344f30 | 2020-05-29 23:26:57 -0700 | [diff] [blame^] | 91 | h.to_topic = self.remote_topic |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 92 | h.to_device_id = self._to_string(to_device_id) |
| 93 | h.proxy_device_id = self._to_string(proxy_device_id) |
| 94 | |
| 95 | if message_id: |
| 96 | h.id = self._to_string(message_id) |
| 97 | else: |
| 98 | h.id = uuid4().hex |
| 99 | |
Scott Baker | 8f14424 | 2020-04-17 13:13:05 -0700 | [diff] [blame] | 100 | h.timestamp.GetCurrentTime() |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 101 | iaMsg = InterAdapterMessage() |
| 102 | iaMsg.header.CopyFrom(h) |
| 103 | iaMsg.body.Pack(msg) |
| 104 | |
Matteo Scandolo | 63efb06 | 2019-11-26 12:14:48 -0700 | [diff] [blame] | 105 | log.debug("sending-inter-adapter-message", type=iaMsg.header.type, from_topic=iaMsg.header.from_topic, |
| 106 | to_topic=iaMsg.header.to_topic, to_device_id=iaMsg.header.to_device_id) |
Chip Boling | 67b674a | 2019-02-08 11:42:18 -0600 | [diff] [blame] | 107 | res = yield self.invoke(rpc="process_inter_adapter_message", |
| 108 | to_topic=iaMsg.header.to_topic, |
| 109 | msg=iaMsg) |
| 110 | returnValue(res) |
| 111 | except Exception as e: |
| 112 | log.exception("error-sending-request", e=e) |