Add data plane commands to az webpubsub

src/webpubsub/azext_webpubsub/_help.py

@@ -32,6 +32,36 @@
     short-summary: Commands to manage Webpubsub network rules.
 """
 
+helps['webpubsub client'] = """
+    type: group
+    short-summary: Commands to manage client connections.
+"""
+
+helps['webpubsub service'] = """
+    type: group
+    short-summary: Commands to manage Webpubsub service.
+"""
+
+helps['webpubsub service connection'] = """
+    type: group
+    short-summary: Commands to manage Webpubsub service connections.
+"""
+
+helps['webpubsub service user'] = """
+    type: group
+    short-summary: Commands to manage Webpubsub service users.
+"""
+
+helps['webpubsub service group'] = """
+    type: group
+    short-summary: Commands to manage Webpubsub service groups.
+"""
+
+helps['webpubsub service permission'] = """
+    type: group
+    short-summary: Commands to manage Webpubsub service permissions.
+"""
+
 helps['webpubsub create'] = """
     type: command
     short-summary: Create a Webpubsub.
@@ -153,3 +183,82 @@
         text: >
             az webpubsub event-handler hub update -n MyWebPubSub -g MyResourceGroup --hub-name MyHub --template url-template='http://host.com user-event-pattern="MyEvent" --template url-template="http://host2.com" system-event-pattern="connect"'
 """
+
+helps['webpubsub client start'] = """
+    type: command
+    short-summary: Start a interactive client connection.
+"""
+
+helps['webpubsub service broadcast'] = """
+    type: command
+    short-summary: Broadcast messages to hub.
+    examples:
+      - name: Send a message to hub
+        text: >
+            az webpubsub service broadcast -n MyWebPubSub -g MyResourceGroup --hub-name MyHub --payload MyPayload
+"""
+
+helps['webpubsub service connection exist'] = """
+    type: command
+    short-summary: Check whether client connection exists
+"""
+
+helps['webpubsub service connection close'] = """
+    type: command
+    short-summary: Close a specific client connection
+"""
+
+helps['webpubsub service connection send'] = """
+    type: command
+    short-summary: Send a message to connection
+"""
+
+helps['webpubsub service group add-connection'] = """
+    type: command
+    short-summary: Add a connection to group
+"""
+
+helps['webpubsub service group remove-connection'] = """
+    type: command
+    short-summary: Remove a connection from group
+"""
+
+helps['webpubsub service group add-user'] = """
+    type: command
+    short-summary: Add a user to group
+"""
+
+helps['webpubsub service group remove-user'] = """
+    type: command
+    short-summary: Remove a user from group
+"""
+
+helps['webpubsub service group send'] = """
+    type: command
+    short-summary: Send a message to group
+"""
+
+helps['webpubsub service user send'] = """
+    type: command
+    short-summary: Send a message to user
+"""
+
+helps['webpubsub service user exist'] = """
+    type: command
+    short-summary: Check if there are any client connections connected for the given user
+"""
+
+helps['webpubsub service permission grant'] = """
+    type: command
+    short-summary: Grant a group permission to the connection.
+"""
+
+helps['webpubsub service permission revoke'] = """
+    type: command
+    short-summary: Revoke a group permission from the connection.
+"""
+
+helps['webpubsub service permission check'] = """
+    type: command
+    short-summary: Check if a connection has permission to the specified group
+"""

src/webpubsub/azext_webpubsub/_params.py

@@ -17,7 +17,7 @@
 
 WEBPUBSUB_KEY_TYPE = ['primary', 'secondary']
 SKU_TYPE = ['Standard_S1', 'Free_F1']
-NETWORK_RULE_TYPE = []
+PERMISSION_TYPE = ['joinLeaveGroup', 'sendToGroup']
 
 
 def load_arguments(self, _):
@@ -56,3 +56,38 @@ def load_arguments(self, _):
 
     with self.argument_context('webpubsub event-handler hub update') as c:
         c.argument('template', action=EventHandlerTemplateUpdateAction, nargs='+', help='Template item for event handler settings. Use key=value pattern to set properties. Supported keys are "url-template", "user-event-pattern", "system-event-pattern".')
+
+    with self.argument_context('webpubsub client') as c:
+        c.argument('hub_name', help='The hub which client connects to')
+
+    with self.argument_context('webpubsub service') as c:
+        c.argument('hub_name', help='The hub to manage.')
+
+    for scope in ['webpubsub service broadcast', 'webpubsub service connection send', 'webpubsub service group send', 'webpubsub service user send']:
+        with self.argument_context(scope) as c:
+            c.argument('payload', help='A string payload to send.')
+
+    for scope in ['webpubsub service connection',
+                  'webpubsub service group add-connection',
+                  'webpubsub service group remove-connection',
+                  'webpubsub service permission grant',
+                  'webpubsub service permission revoke',
+                  'webpubsub service permission check']:
+        with self.argument_context(scope) as c:
+            c.argument('connection_id', help='The connection id.')
+
+    for scope in ['webpubsub service group',
+                  'webpubsub service permission grant',
+                  'webpubsub service permission revoke',
+                  'webpubsub service permission check']:
+        with self.argument_context(scope) as c:
+            c.argument('group_name', help='The group name.')
+
+    for scope in ['webpubsub service group add-user',
+                  'webpubsub service group remove-user',
+                  'webpubsub service user']:
+        with self.argument_context(scope) as c:
+            c.argument('user_id', help='The user id.')
+
+    with self.argument_context('webpubsub service permission') as c:
+        c.argument('permission', arg_type=get_enum_type(PERMISSION_TYPE), help='The permission')

src/webpubsub/azext_webpubsub/client.py

@@ -0,0 +1,124 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+# pylint: disable=line-too-long
+
+import asyncio
+import sys
+import threading
+import json
+import websockets
+from .vendored_sdks.azure_messaging_webpubsubservice import (
+    build_authentication_token
+)
+
+
+HELP_MESSAGE = """
+----------Usage-----------
+help                                : Print help messages
+joingroup <group-name>              : Join the connection to group
+leavegroup <group-name>             : Leave the connection from group
+sendtogroup <group-name> <message>  : Send message to group
+event <event-name> <message>        : Send event to event handler
+--------------------------
+        """
+
+
+async def connect(url):
+    async with websockets.connect(url, subprotocols=['json.webpubsub.azure.v1']) as ws:
+
+        eprint(HELP_MESSAGE)
+        publisher = Publisher(ws)
+        publisher.daemon = True
+        publisher.start()
+        while True:
+            print(await ws.recv())
+
+
+def start_client(client, resource_group_name, webpubsub_name, hub_name):
+    keys = client.list_keys(resource_group_name, webpubsub_name)
+    connection_string = keys.primary_connection_string
+    token = build_authentication_token(connection_string, hub_name, roles=['webpubsub.sendToGroup', 'webpubsub.joinLeaveGroup'])
+    asyncio.get_event_loop().run_until_complete(connect(token['url']))
+
+
+def eprint(*args, **kwargs):
+    print(*args, file=sys.stderr, **kwargs)
+
+
+class Publisher(threading.Thread):
+    def __init__(self, ws):
+        threading.Thread.__init__(self)
+        self.ws = ws
+        self.id = 0
+
+    def run(self):
+        new_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(new_loop)
+        while True:
+            input_line = sys.stdin.readline().strip()
+            asyncio.get_event_loop().run_until_complete(self._parse(input_line))
+
+    def join(self, timeout=None):
+        super().join()
+
+    async def _parse(self, input_line):
+        if input_line:
+            if input_line.strip() == 'help':
+                eprint(HELP_MESSAGE)
+                return
+
+            arr = input_line.split(maxsplit=1)
+            if len(arr) != 2:
+                eprint('Invalid input "{}", use help to show usage'.format(input_line))
+                return
+
+            command = arr[0]
+            if command.lower() == 'joingroup':
+                group = arr[1]
+                payload = json.dumps({
+                    'type': 'joinGroup',
+                    'group': group,
+                    'ackId': self._get_ack_id()
+                })
+                await self.ws.send(payload)
+
+            elif command.lower() == 'leavegroup':
+                group = arr[1]
+                payload = json.dumps({
+                    'type': 'leaveGroup',
+                    'group': group,
+                    'ackId': self._get_ack_id()
+                })
+                await self.ws.send(payload)
+
+            elif command.lower() == 'sendtogroup':
+                arr = arr[1].split(maxsplit=1)
+                group = arr[0]
+                data = arr[1]
+                payload = json.dumps({
+                    'type': 'sendToGroup',
+                    'group': group,
+                    'data': data,
+                    'ackId': self._get_ack_id()
+                })
+                await self.ws.send(payload)
+
+            elif command.lower() == 'event':
+                arr = arr[1].split(maxsplit=1)
+                event = arr[0]
+                data = arr[1]
+                payload = json.dumps({
+                    'type': 'event',
+                    'event': event,
+                    'data': data
+                })
+                await self.ws.send(payload)
+
+            else:
+                eprint('Invalid input "{}", use help to show usage'.format(input_line))
+
+    def _get_ack_id(self):
+        self.id = self.id + 1
+        return self.id

src/webpubsub/azext_webpubsub/commands.py

@@ -31,6 +31,16 @@ def load_command_table(self, _):
         client_factory=cf_webpubsub
     )
 
+    webpubsub_client_utils = CliCommandType(
+        operations_tmpl='azext_webpubsub.client#{}',
+        client_factory=cf_webpubsub
+    )
+
+    webpubsub_service_utils = CliCommandType(
+        operations_tmpl='azext_webpubsub.service#{}',
+        client_factory=cf_webpubsub
+    )
+
     with self.command_group('webpubsub', webpubsub_general_utils, is_preview=True) as g:
         g.command('create', 'webpubsub_create')
         g.command('delete', 'webpubsub_delete')
@@ -57,3 +67,30 @@ def load_command_table(self, _):
     with self.command_group('webpubsub event-handler hub', webpubsub_eventhandler_utils) as g:
         g.command('remove', 'event_handler_hub_remove')
         g.command('update', 'event_handler_hub_update')
+
+    with self.command_group('webpubsub client', webpubsub_client_utils) as g:
+        g.command('start', 'start_client')
+
+    with self.command_group('webpubsub service', webpubsub_service_utils) as g:
+        g.command('broadcast', 'broadcast')
+
+    with self.command_group('webpubsub service connection', webpubsub_service_utils) as g:
+        g.command('exist', 'check_connection_exists')
+        g.command('close', 'close_connection')
+        g.command('send', 'send_connection')
+
+    with self.command_group('webpubsub service group', webpubsub_service_utils) as g:
+        g.command('add-connection', 'add_connection_to_group')
+        g.command('remove-connection', 'remove_connection_from_group')
+        g.command('send', 'send_group')
+        g.command('add-user', 'add_user_to_group')
+        g.command('remove-user', 'remove_user_from_group')
+
+    with self.command_group('webpubsub service user', webpubsub_service_utils) as g:
+        g.command('send', 'send_user')
+        g.command('exist', 'check_user_exists')
+
+    with self.command_group('webpubsub service permission', webpubsub_service_utils) as g:
+        g.command('grant', 'grant_permission')
+        g.command('revoke', 'revoke_permission')
+        g.command('check', 'check_permission')