Feat conn string builders

common/file_rotate.py

@@ -7,3 +7,6 @@ class FileRotate(Enum):
     DAILY = 2
     WEEKLY = 3
     MONTHLY = 4
+
+    def __str__(self):
+        return "%s" % self._name_

common/lookup_table.py

@@ -105,8 +105,9 @@ def clear(self):
     def get_count(self):
         return len(self.__items)
 
-    def __is_valid_integer(self, value: str) -> bool:
-        if self.__value_is_valid(value):
+    @classmethod
+    def __is_valid_integer(cls, value: str) -> bool:
+        if cls.__value_is_valid(value):
             try:
                 int(value)
                 return True
@@ -145,30 +146,31 @@ def get_size_value(self, key: str, default_value: int) -> int:
 
         return result
 
-    def size_to_int(self, value: str, default_value: int) -> int:
+    @classmethod
+    def size_to_int(cls, value: str, default_value: int) -> int:
         if not isinstance(value, str):
             raise TypeError("Value must be a string")
         if not isinstance(default_value, int):
             raise TypeError("Default value must be an int")
 
         result = default_value
-        factor = self.__KB_FACTOR
+        factor = cls.__KB_FACTOR
         value = value.strip()
 
         if len(value) >= 2:
             unit = value[-2:].lower()
 
-            if self.__is_valid_size_unit(unit):
+            if cls.__is_valid_size_unit(unit):
                 value = value[:-2].strip()
 
                 if unit == "kb":
-                    factor = self.__KB_FACTOR
+                    factor = cls.__KB_FACTOR
                 elif unit == "mb":
-                    factor = self.__MB_FACTOR
+                    factor = cls.__MB_FACTOR
                 elif unit == "gb":
-                    factor = self.__GB_FACTOR
+                    factor = cls.__GB_FACTOR
 
-        if self.__is_valid_integer(value):
+        if cls.__is_valid_integer(value):
             try:
                 result = factor * int(value)
             except ValueError:

common/pattern_parser.py

@@ -68,14 +68,14 @@ def _next(self) -> typing.Optional[Token]:
             is_variable = False
             pos: int = self._position
 
-            if self._pattern[pos: pos + 2] == "/%":
+            if self._pattern[pos] == "$":
                 is_variable = True
-                pos += 2
+                pos += 1
 
-            while pos + 1 < length:
-                if self._pattern[pos: pos + 2] == "/%":
+            while pos < length:
+                if self._pattern[pos] == "$":
                     if is_variable:
-                        pos += 2
+                        pos += 1
                     break
                 pos += 1
 

common/tokens/token_factory.py

@@ -4,17 +4,17 @@
 
 class TokenFactory:
     tokens = {
-        "/%appname/%": AppNameToken,
-        "/%session/%": SessionToken,
-        "/%hostname/%": HostNameToken,
-        "/%title/%": TitleToken,
-        "/%timestamp/%": TimestampToken,
-        "/%level/%": LevelToken,
-        "/%color/%": ColorToken,
-        "/%logentrytype/%": LogEntryTypeToken,
-        "/%viewerid/%": ViewerIdToken,
-        "/%thread/%": ThreadIdToken,
-        "/%process/%": ProcessIdToken,
+        "$appname$": AppNameToken,
+        "$session$": SessionToken,
+        "$hostname$": HostNameToken,
+        "$title$": TitleToken,
+        "$timestamp$": TimestampToken,
+        "$level$": LevelToken,
+        "$color$": ColorToken,
+        "$logentrytype$": LogEntryTypeToken,
+        "$viewerid$": ViewerIdToken,
+        "$thread$": ThreadIdToken,
+        "$process$": ProcessIdToken,
     }
 
     @staticmethod
@@ -37,30 +37,29 @@ def get_token(cls, value: str) -> Token:
         if length <= 2:
             return cls._create_literal(value)
 
-        if value[:2] != "/%" or value[-2:] != "/%":
+        if value[0] != "$" or value[-1] != "$":
             return cls._create_literal(value)
 
         original = value
         options = ""
 
-        # extract the token options: /%token{options}/%
-        if value[-3] == "}":
+        # extract the token options: $token{options}$
+        if value[-2] == "}":
             idx = value.find("{")
 
             if idx > -1:
                 idx += 1
-                options = value[idx: -3]
-                value = value[:idx - 1] + value[-2:]
-                length = len(value)
+                options = value[idx: -2]
+                value = value[:idx - 1] + value[-1]
 
         width = ""
         idx = value.find(",")
 
-        # extract the token width: /%token, width/%
+        # extract the token width: $token, width$
         if idx > -1:
             idx += 1
-            width = value[idx: -2]
-            value = value[: idx - 1] + value[length - 2:]
+            width = value[idx: -1]
+            value = value[: idx - 1] + value[-1]
 
         value = value.lower()
         impl = cls.tokens.get(value)

connections/builders/__init__.py

@@ -1 +1 @@
-from connections.builders.connections_builder import ConnectionsBuilder
+from connections.builders.connection_string_builder import ConnectionStringBuilder

connections/builders/cloud_protocol_connection_string_builder.py

@@ -0,0 +1,70 @@
+import collections
+from typing import TYPE_CHECKING, TypeVar
+
+from common.file_rotate import FileRotate
+from common.lookup_table import LookupTable
+from protocols.cloud.cloud_protocol import CloudProtocol
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.tcp_protocol_connection_string_builder import TcpProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="CloudProtocolConnectionStringBuilder")
+
+
+class CloudProtocolConnectionStringBuilder(TcpProtocolConnectionStringBuilder):
+    _custom_labels: collections.OrderedDict = collections.OrderedDict()
+
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def end_protocol(self) -> "ConnectionStringBuilder":
+        self._parent.cb.add_option("customlabels", CloudProtocol.compose_custom_labels_string(self._custom_labels))
+        return super().end_protocol()
+
+    def set_write_key(self, write_key: str) -> Self:
+        self._parent.cb.add_option("writekey", write_key)
+        return self
+
+    def add_custom_label(self, key: str, value: str) -> Self:
+        self._custom_labels[key] = value
+        return self
+
+    def set_region(self, region: str) -> Self:
+        self._parent.cb.add_option("region", region)
+        return self
+
+    def set_chunking_enabled(self, enabled: bool) -> Self:
+        self._parent.cb.add_option("chunking.enabled", enabled)
+        return self
+
+    def set_chunking_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("chunking.maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_chunking_max_age_ms(self, max_age: int) -> Self:
+        self._parent.cb.add_option("chunking.maxagems", max_age)
+        return self
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_rotate(self, rotate: FileRotate) -> Self:
+        self._parent.cb.add_option("rotate", rotate)
+        return self
+
+    def set_tls_enabled(self, tls_enabled: bool) -> Self:
+        self._parent.cb.add_option("tls.enabled", tls_enabled)
+        return self
+
+    def set_tls_certificate_location(self, tls_certificate_location: str) -> Self:
+        self._parent.cb.add_option("tls.certificate.location", tls_certificate_location)
+        return self
+
+    def set_tls_certificate_filepath(self, tls_certificate_filepath: str) -> Self:
+        self._parent.cb.add_option("tls.certificate.filepath", tls_certificate_filepath)
+        return self

connections/builders/connection_string_builder.py

@@ -0,0 +1,50 @@
+from connections.builders.cloud_protocol_connection_string_builder import CloudProtocolConnectionStringBuilder
+from connections.builders.file_protocol_connection_string_builder import FileProtocolConnectionStringBuilder
+from connections.builders.memory_protocol_connection_string_builder import MemoryProtocolConnectionStringBuilder
+from connections.builders.pipe_protocol_connection_string_builder import PipeProtocolConnectionStringBuilder
+from connections.builders.tcp_protocol_connection_string_builder import TcpProtocolConnectionStringBuilder
+from connections.builders.text_protocol_connection_string_builder import TextProtocolConnectionStringBuilder
+from connections.connections_builder import ConnectionsBuilder
+
+
+class ConnectionStringBuilder:
+    _protocols = {
+        "pipe": PipeProtocolConnectionStringBuilder,
+        "file": FileProtocolConnectionStringBuilder,
+        "mem": MemoryProtocolConnectionStringBuilder,
+        "tcp": TcpProtocolConnectionStringBuilder,
+        "text": TextProtocolConnectionStringBuilder,
+        "cloud": CloudProtocolConnectionStringBuilder,
+    }
+
+    def __init__(self):
+        self._cb = ConnectionsBuilder()
+
+    @property
+    def cb(self):
+        return self._cb
+
+    def _add_protocol(self, protocol_name: str):
+        self._cb.begin_protocol(protocol_name)
+        return self._protocols.get(protocol_name)(self)
+
+    def add_pipe_protocol(self) -> PipeProtocolConnectionStringBuilder:
+        return self._add_protocol("pipe")
+
+    def add_file_protocol(self) -> FileProtocolConnectionStringBuilder:
+        return self._add_protocol("file")
+
+    def add_memory_protocol(self) -> MemoryProtocolConnectionStringBuilder:
+        return self._add_protocol("mem")
+
+    def add_tcp_protocol(self) -> TcpProtocolConnectionStringBuilder:
+        return self._add_protocol("tcp")
+
+    def add_text_protocol(self) -> TextProtocolConnectionStringBuilder:
+        return self._add_protocol("text")
+
+    def add_cloud_protocol(self) -> CloudProtocolConnectionStringBuilder:
+        return self._add_protocol("cloud")
+
+    def build(self) -> str:
+        return self._cb.get_connections()

connections/builders/file_protocol_connection_string_builder.py

@@ -0,0 +1,51 @@
+from typing import TYPE_CHECKING, TypeVar
+
+from common.file_rotate import FileRotate
+from common.lookup_table import LookupTable
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="FileProtocolConnectionStringBuilder")
+
+
+class FileProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_filename(self, filename: str) -> Self:
+        self._parent.cb.add_option("filename", filename)
+        return self
+
+    def set_append(self, append: bool) -> Self:
+        self._parent.cb.add_option("append", append)
+        return self
+
+    def set_buffer(self, buffer: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(buffer, 0)
+        self._parent.cb.add_option("buffer", int(size_in_bytes / 1024))
+        return self
+
+    def set_rotate(self, rotate: FileRotate) -> Self:
+        self._parent.cb.add_option("rotate", rotate)
+        return self
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_max_parts(self, max_parts: int) -> Self:
+        self._parent.cb.add_option("maxparts", max_parts)
+        return self
+
+    def set_key(self, key: str) -> Self:
+        self._parent.cb.add_option("key", key)
+        return self
+
+    def set_encrypt(self, encrypt: bool) -> Self:
+        self._parent.cb.add_option("encrypt", encrypt)
+        return self

connections/builders/memory_protocol_connection_string_builder.py

@@ -0,0 +1,32 @@
+from typing import TYPE_CHECKING, TypeVar
+
+from common.lookup_table import LookupTable
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="MemoryProtocolConnectionStringBuilder")
+
+
+class MemoryProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_max_size(self, max_size: str) -> Self:
+        size_in_bytes = LookupTable.size_to_int(max_size, 0)
+        self._parent.cb.add_option("maxsize", int(size_in_bytes / 1024))
+        return self
+
+    def set_as_text(self, as_text: bool) -> Self:
+        self._parent.cb.add_option("astext", as_text)
+        return self
+
+    def set_indent(self, indent: bool) -> Self:
+        self._parent.cb.add_option("indent", indent)
+        return self
+
+    def set_pattern(self, pattern: str) -> Self:
+        self._parent.cb.add_option("pattern", pattern)
+        return self

connections/builders/pipe_protocol_connection_string_builder.py

@@ -0,0 +1,17 @@
+from typing import TYPE_CHECKING, TypeVar
+
+if TYPE_CHECKING:
+    from connection_string_builder import ConnectionStringBuilder
+
+from connections.builders.protocol_connection_string_builder import ProtocolConnectionStringBuilder
+
+Self = TypeVar("Self", bound="PipeProtocolConnectionStringBuilder")
+
+
+class PipeProtocolConnectionStringBuilder(ProtocolConnectionStringBuilder):
+    def __init__(self, parent: "ConnectionStringBuilder"):
+        super().__init__(parent)
+
+    def set_pipe_name(self, pipe_name: str) -> Self:
+        self._parent.cb.add_option("pipename", pipe_name)
+        return self