# Path: vnpy/rpc/common.py
# HEARTBEAT_TOPIC: str = "heartbeat"
# 
# HEARTBEAT_INTERVAL: int = 10


import threading
import traceback
import zmq
from time import time
from typing import Any, Callable, Dict
from .common import HEARTBEAT_TOPIC, HEARTBEAT_INTERVAL
            except Exception as e:  # noqa
                rep = [False, traceback.format_exc()]

            # send callable response by Reply socket
            self._socket_rep.send_pyobj(rep)

        # Unbind socket address
        self._socket_pub.unbind(self._socket_pub.LAST_ENDPOINT)
        self._socket_rep.unbind(self._socket_rep.LAST_ENDPOINT)

    def publish(self, topic: str, data: Any) -> None:
        """
        Publish data
        """
        with self._lock:
            self._socket_pub.send_pyobj([topic, data])

    def register(self, func: Callable) -> None:
        """
        Register function
        """
        self._functions[func.__name__] = func

    def check_heartbeat(self) -> None:
        """
        Check whether it is required to send heartbeat.
        """
        now: float = time()
        if now >= self._heartbeat_at:
            # Publish heartbeat

# Path: vnpy/rpc/common.py
# HEARTBEAT_TOPIC: str = "heartbeat"
# 
# HEARTBEAT_INTERVAL: int = 10


# Path: vnpy/rpc/server.py
import threading
import traceback
import zmq
from time import time
from typing import Any, Callable, Dict
from .common import HEARTBEAT_TOPIC, HEARTBEAT_INTERVAL

            except Exception as e:  # noqa
                rep = [False, traceback.format_exc()]

            # send callable response by Reply socket
            self._socket_rep.send_pyobj(rep)

        # Unbind socket address
        self._socket_pub.unbind(self._socket_pub.LAST_ENDPOINT)
        self._socket_rep.unbind(self._socket_rep.LAST_ENDPOINT)

    def publish(self, topic: str, data: Any) -> None:
        """
        Publish data
        """
        with self._lock:
            self._socket_pub.send_pyobj([topic, data])

    def register(self, func: Callable) -> None:
        """
        Register function
        """
        self._functions[func.__name__] = func

    def check_heartbeat(self) -> None:
        """
        Check whether it is required to send heartbeat.
        """
        now: float = time()
        if now >= self._heartbeat_at:
            # Publish heartbeat

            self.publish(HEARTBEAT_TOPIC, now)