dimp 2.4.1

Decentralized Instant Messaging Protocol

Decentralized Instant Messaging Protocol (Python)

Dependencies

Name	Version	Description
Ming Ke Ming (名可名)		Decentralized User Identity Authentication
Dao Ke Dao (道可道)		Universal Message Module

Examples

Extends Command

• Handshake Command Protocol 0. (C-S) handshake start
  1. (S-C) handshake again with new session
  2. (C-S) handshake restart with new session
  3. (S-C) handshake success

from abc import ABC, abstractmethod
from enum import IntEnum
from typing import Optional, Dict

from dimp import *


class HandshakeState(IntEnum):
    Start = 0    # C -> S, without session key(or session expired)
    Again = 1    # S -> C, with new session key
    Restart = 2  # C -> S, with new session key
    Success = 3  # S -> C, handshake accepted


def handshake_state(title: str, session: str = None) -> HandshakeState:
    # Server -> Client
    if title == 'DIM!':  # or title == 'OK!':
        return HandshakeState.Success
    if title == 'DIM?':
        return HandshakeState.Again
    # Client -> Server: "Hello world!"
    if session is None or len(session) == 0:
        return HandshakeState.Start
    else:
        return HandshakeState.Restart


class HandshakeCommand(Command, ABC):
    """
        Handshake Command
        ~~~~~~~~~~~~~~~~~

        data format: {
            "type" : i2s(0x88),
            "sn"   : 12345,

            "command" : "handshake",    // command name
            "title"   : "Hello world!", // "DIM?", "DIM!"
            "session" : "{SESSION_ID}", // session key
        }
    """
    HANDSHAKE = 'handshake'

    @property
    @abstractmethod
    def title(self) -> str:
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.title getter'
        )

    @property
    @abstractmethod
    def session(self) -> Optional[str]:
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.session getter'
        )

    @property
    @abstractmethod
    def state(self) -> HandshakeState:
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.state getter'
        )

    #
    #   Factories
    #

    @classmethod
    def offer(cls, session: str = None) -> Command:
        """
        Create client-station handshake offer

        :param session: Old session key
        :return: HandshakeCommand object
        """
        return BaseHandshakeCommand(title='Hello world!', session=session)

    @classmethod
    def ask(cls, session: str) -> Command:
        """
        Create station-client handshake again with new session

        :param session: New session key
        :return: HandshakeCommand object
        """
        return BaseHandshakeCommand(title='DIM?', session=session)

    @classmethod
    def accepted(cls, session: str = None) -> Command:
        """
        Create station-client handshake success notice

        :return: HandshakeCommand object
        """
        return BaseHandshakeCommand(title='DIM!', session=session)

    start = offer       # (1. C->S) first handshake, without session
    again = ask         # (2. S->C) ask client to handshake with new session key
    restart = offer     # (3. C->S) handshake with new session key
    success = accepted  # (4. S->C) notice the client that handshake accepted


class BaseHandshakeCommand(BaseCommand, HandshakeCommand):

    def __init__(self, content: Dict = None, title: str = None, session: str = None):
        if content is None:
            # 1. new command with title & session key
            assert title is not None, 'handshake command error: %s' % session
            cmd = self.HANDSHAKE
            super().__init__(cmd=cmd)
            self['title'] = title
            self['message'] = title  # TODO: remove after all clients upgraded
            if session is not None:
                self['session'] = session
        else:
            # 2. command info from network
            assert title is None and session is None, 'params error: %s, %s, %s' % (content, title, session)
            super().__init__(content)

    @property
    def title(self) -> str:
        return self.get_str(key='title', default='')

    @property
    def session(self) -> Optional[str]:
        return self.get_str(key='session')

    @property
    def state(self) -> HandshakeState:
        return handshake_state(title=self.title, session=self.session)

Extends Content

from abc import ABC, abstractmethod

from dimp import *


class AppContent(Content, ABC):
    """
        Content for Application 0nly
        ~~~~~~~~~~~~~~~~~~~~~~~~~~~~

        data format: {
            "type" : i2s(0xA0),
            "sn"   : 12345,

            "app"   : "{APP_ID}",  // application (e.g.: "chat.dim.sechat")
            "extra" : info         // action parameters
        }
    """

    @property
    @abstractmethod
    def application(self) -> str:
        """ App ID """
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.application getter'
        )


class CustomizedContent(Content, ABC):
    """
        Application Customized message
        ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

        data format: {
            "type" : i2s(0xCC),
            "sn"   : 12345,

            "app"   : "{APP_ID}",  // application (e.g.: "chat.dim.sechat")
            "mod"   : "{MODULE}",  // module name (e.g.: "drift_bottle")
            "act"   : "{ACTION}",  // action name (e.g.: "throw")
            "extra" : info         // action parameters
        }
    """

    @property
    @abstractmethod
    def module(self) -> str:
        """ Module Name """
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.module getter'
        )

    @property
    @abstractmethod
    def action(self) -> str:
        """ Action Name """
        raise NotImplementedError(
            f'Not implemented: {type(self).__module__}.{type(self).__name__}.action getter'
        )

    #
    #   Factory method
    #
    @classmethod
    def create(cls, app: str, mod: str, act: str):
        return AppCustomizedContent(app=app, mod=mod, act=act)

from typing import Dict

from dimp import *


class AppCustomizedContent(BaseContent, AppContent, CustomizedContent):

    def __init__(self, content: Dict = None,
                 msg_type: str = None,
                 app: str = None, mod: str = None, act: str = None):
        if content is None:
            # 1. new content with type, application, module & action
            assert app is not None and mod is not None and act is not None, \
                'customized content error: %s, %s, %s, %s' % (msg_type, app, mod, act)
            if msg_type is None:
                msg_type = ContentType.CUSTOMIZED
            super().__init__(None, msg_type)
            self['app'] = app
            self['mod'] = mod
            self['act'] = act
        else:
            # 2. content info from network
            assert msg_type is None and app is None and mod is None and act is None, \
                'params error: %s, %s, %s, %s, %s' % (content, msg_type, app, mod, act)
            super().__init__(content)

    @property  # Override
    def application(self) -> str:
        return self.get_str(key='app', default='')

    @property  # Override
    def module(self) -> str:
        return self.get_str(key='mod', default='')

    @property  # Override
    def action(self) -> str:
        return self.get_str(key='act', default='')

Extends ID Address

• Examples in dim plugins

---

Copyright © 2018-2026 Albert Moky