Commit d0c0b2ff authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Context component: 

- Skip NATS client port from linkerd
- Add NATS backend logging messages
parent b9d3eed0

manifests/contextservice.yaml

+6 −4
Original line number Original line Diff line number Diff line
@@ -23,6 +23,8 @@ spec: 
  #replicas: 1

  #replicas: 1

  template:

  template:

    metadata:

    metadata:

      annotations:

        config.linkerd.io/skip-outbound-ports: "4222"

      labels:

      labels:

        app: contextservice

        app: contextservice

    spec:

    spec:
@@ -52,11 +54,11 @@ spec: 
            command: ["/bin/grpc_health_probe", "-addr=:1010"]

            command: ["/bin/grpc_health_probe", "-addr=:1010"]

        resources:

        resources:

          requests:

          requests:

            cpu: 75m

            cpu: 250m

            memory: 64Mi

          limits:

            cpu: 100m

            memory: 128Mi

            memory: 128Mi

          limits:

            cpu: 1000m

            memory: 1024Mi

---

---

apiVersion: v1

apiVersion: v1

kind: Service

kind: Service

src/common/message_broker/backend/nats/NatsBackendThread.py

+8 −1
Original line number Original line Diff line number Diff line
@@ -12,10 +12,12 @@ 
# See the License for the specific language governing permissions and

# See the License for the specific language governing permissions and

# limitations under the License.

# limitations under the License.





import asyncio, nats, nats.errors, queue, threading

import asyncio, logging, nats, nats.errors, queue, threading

from typing import List

from typing import List

from common.message_broker.Message import Message

from common.message_broker.Message import Message





LOGGER = logging.getLogger(__name__)



class NatsBackendThread(threading.Thread):

class NatsBackendThread(threading.Thread):

    def __init__(self, nats_uri : str) -> None:

    def __init__(self, nats_uri : str) -> None:

        self._nats_uri = nats_uri

        self._nats_uri = nats_uri
@@ -32,7 +34,9 @@ class NatsBackendThread(threading.Thread): 
        self._tasks_terminated.set()

        self._tasks_terminated.set()





    async def _run_publisher(self) -> None:

    async def _run_publisher(self) -> None:

        LOGGER.info('[_run_publisher] NATS URI: {:s}'.format(str(self._nats_uri)))

        client = await nats.connect(servers=[self._nats_uri])

        client = await nats.connect(servers=[self._nats_uri])

        LOGGER.info('[_run_publisher] Connected!')

        while not self._terminate.is_set():

        while not self._terminate.is_set():

            try:

            try:

                message : Message = await self._publish_queue.get()

                message : Message = await self._publish_queue.get()
@@ -47,8 +51,11 @@ class NatsBackendThread(threading.Thread): 
    async def _run_subscriber(

    async def _run_subscriber(

        self, topic_name : str, timeout : float, out_queue : queue.Queue[Message], unsubscribe : threading.Event

        self, topic_name : str, timeout : float, out_queue : queue.Queue[Message], unsubscribe : threading.Event

    ) -> None:

    ) -> None:

        LOGGER.info('[_run_subscriber] NATS URI: {:s}'.format(str(self._nats_uri)))

        client = await nats.connect(servers=[self._nats_uri])

        client = await nats.connect(servers=[self._nats_uri])

        LOGGER.info('[_run_subscriber] Connected!')

        subscription = await client.subscribe(topic_name)

        subscription = await client.subscribe(topic_name)

        LOGGER.info('[_run_subscriber] Subscribed!')

        while not self._terminate.is_set() and not unsubscribe.is_set():

        while not self._terminate.is_set() and not unsubscribe.is_set():

            try:

            try:

                message = await subscription.next_msg(timeout)

                message = await subscription.next_msg(timeout)