feat: subscribe custom events filter

Author: pozil

src/main/default/classes/StreamingMonitorController.cls

@@ -3,7 +3,7 @@ public abstract class StreamingMonitorController {
     public final static String EVT_GENERIC = 'GenericEvent';
     public final static String EVT_STD_PLATFORM_EVENT = 'StandardPlatformEvent';
     public final static String EVT_PLATFORM_EVENT = 'PlatformEvent';
-    public final static String EVT_CDC_STANDARD = 'ChangeDataCaptureEvent';
+    public final static String EVT_CDC = 'ChangeDataCaptureEvent';
     public final static String EVT_CUSTOM_CHANNEL_CDC = 'CustomChannelCDC';
     public final static String EVT_CUSTOM_CHANNEL_PE = 'CustomChannelPE';
     public final static String EVT_MONITORING = 'MonitoringEvent';
@@ -56,7 +56,7 @@ public abstract class StreamingMonitorController {
             EVT_GENERIC => getGenericEventChannels(),
             EVT_STD_PLATFORM_EVENT => getStandardPlatformEventChannels(),
             EVT_PLATFORM_EVENT => getPlatformEventChannels(),
-            EVT_CDC_STANDARD => getChangeDataCaptureEventChannels(),
+            EVT_CDC => getChangeDataCaptureEventChannels(),
             EVT_CUSTOM_CHANNEL_CDC => new List<ComboBoxItem>(),
             EVT_CUSTOM_CHANNEL_PE => new List<ComboBoxItem>(),
             EVT_MONITORING => getEventMonitoringChannels()
@@ -307,7 +307,7 @@ public abstract class StreamingMonitorController {
         public void execute(QueueableContext context) {
             String restAPIURL =
                 URL.getOrgDomainUrl().toExternalForm() +
-                '/services/data/v60.0/sobjects/StreamingChannel/' +
+                '/services/data/v62.0/sobjects/StreamingChannel/' +
                 channelId +
                 '/push';
             HttpRequest httpRequest = new HttpRequest();

src/main/default/lwc/actions/actions.js

@@ -6,10 +6,12 @@ import {
     EVT_GENERIC,
     EVT_STD_PLATFORM_EVENT,
     EVT_PLATFORM_EVENT,
-    EVT_CDC_STANDARD,
+    EVT_CDC,
     EVT_CUSTOM_CHANNEL_CDC,
     EVT_CUSTOM_CHANNEL_PE,
     EVT_MONITORING,
+    FILTER_ALL,
+    FILTER_CUSTOM,
     getChannelPrefix
 } from 'c/streamingUtility';
 
@@ -27,6 +29,7 @@ export default class Actions extends LightningElement {
     @api action = 'subscribeAll';
     @api channels = [];
 
+    subAllFilter = FILTER_ALL;
     subAllReplay = '-1';
 
     subEventType;
@@ -60,6 +63,7 @@ export default class Actions extends LightningElement {
     handleSubscribeAll() {
         const subscribeEvent = new CustomEvent('subscribeall', {
             detail: {
+                filter: this.subAllFilter,
                 replayId: this.subAllReplay
             }
         });
@@ -301,7 +305,7 @@ export default class Actions extends LightningElement {
     }
 
     get isStandardCDCReg() {
-        return this.regEventType === EVT_CDC_STANDARD;
+        return this.regEventType === EVT_CDC;
     }
 
     get isCustomChannelCDCReg() {
@@ -318,7 +322,7 @@ export default class Actions extends LightningElement {
 
     get isCDCSub() {
         return (
-            this.subEventType === EVT_CDC_STANDARD ||
+            this.subEventType === EVT_CDC ||
             this.subEventType === EVT_CUSTOM_CHANNEL_CDC
         );
     }
@@ -332,4 +336,11 @@ export default class Actions extends LightningElement {
             ? 'Plain string payload'
             : 'JSON formatted payload with strings delimited by double quotes';
     }
+
+    get subAllFilterOptions() {
+        return [
+            { label: 'All events', value: FILTER_ALL },
+            { label: 'Custom events only', value: FILTER_CUSTOM }
+        ];
+    }
 }

src/main/default/lwc/actions/subscribeAll.html

@@ -1,11 +1,19 @@
 <template>
-    <lightning-card title="Subscribe to all streaming channels">
+    <lightning-card title="Subscribe to multiple streaming channels">
         <div class="slds-p-horizontal_large">
             <p class="slds-m-vertical_small">
                 Custom Change Data Capture channels cannot be automatically
                 discovered. Supported channels: PushTopic, Generic, Platform
                 events, Change Data Capture events and Monitoring events.
             </p>
+            <lightning-combobox
+                label="Filter"
+                name="subAllFilter"
+                value={subAllFilter}
+                onchange={handleValueChange}
+                options={subAllFilterOptions}
+            >
+            </lightning-combobox>
             <lightning-combobox
                 label="Replay option"
                 name="subAllReplay"
@@ -16,7 +24,7 @@
             <div class="slds-align_absolute-center slds-m-top_medium">
                 <lightning-button
                     variant="brand"
-                    label="Subscribe to all"
+                    label="Subscribe"
                     onclick={handleSubscribeAll}
                 ></lightning-button>
             </div>

src/main/default/lwc/sidebar/sidebar.html

@@ -34,7 +34,7 @@
                 ></lightning-vertical-navigation-item-icon>
                 <lightning-vertical-navigation-section label="Actions">
                     <lightning-vertical-navigation-item
-                        label="Subscribe to all channels"
+                        label="Subscribe to channels"
                         name="subscribeAll"
                     ></lightning-vertical-navigation-item>
                     <lightning-vertical-navigation-item

src/main/default/lwc/streamingMonitor/streamingMonitor.js

@@ -12,16 +12,19 @@ import getAllEventChannels from '@salesforce/apex/StreamingMonitorController.get
 import publishStreamingEvent from '@salesforce/apex/StreamingMonitorController.publishStreamingEvent';
 import {
     EVENT_TYPES,
-    EVT_CDC_STANDARD,
+    EVT_CDC,
     CHANNEL_ALL_CDC,
+    FILTER_CUSTOM,
+    FILTER_ALL,
     isCDCChannel,
     getChannelPrefix,
     normalizeEvent,
-    channelSort
+    channelSort,
+    isCustomChannel
 } from 'c/streamingUtility';
 
 const RERENDER_DELAY = 200;
-const IGNORE_SUBCRIBE_ERRORS_DELAY = 3000;
+const IGNORE_SUBCRIBE_ERRORS_DELAY = 4000;
 
 const VIEW_MONITOR = 'monitor';
 const VIEW_SUBSCRIBE_ALL = 'subscribeAll';
@@ -40,20 +43,17 @@ export default class StreamingMonitor extends LightningElement {
     eventsElement;
     rerenderTimeout;
 
-    connectedCallback() {
+    async connectedCallback() {
         setDebugFlag(true);
 
         onError((error) => this.handleStreamingError(error));
 
-        getAllEventChannels()
-            .then((allChannels) => {
-                this.channels = allChannels;
-            })
-            .catch((error) => {
-                console.error(JSON.stringify(error));
-                throw new Error('Failed to retrieve streaming channels');
-            });
-
+        try {
+            this.channels = await getAllEventChannels();
+        } catch (error) {
+            console.error(JSON.stringify(error));
+            throw new Error('Failed to retrieve streaming channels');
+        }
         window.addEventListener('resize', this.handleWindowResize.bind(this));
     }
 
@@ -137,53 +137,81 @@ export default class StreamingMonitor extends LightningElement {
         }
     }
 
-    handleSubscribeAll(event) {
-        console.log(`Subscribing to all streaming events`);
-        const { replayId } = event.detail;
-
-        // Temporarily ignore subscribe errors while subscribing to all events
-        this.ignoreSubscribeErrors = true;
-        setTimeout(() => {
-            this.ignoreSubscribeErrors = false;
-        }, IGNORE_SUBCRIBE_ERRORS_DELAY);
+    async handleSubscribeAll(event) {
+        const { replayId, filter } = event.detail;
+        console.log(
+            `Subscribing to multiple streaming channels with filter ${filter} and replay ID ${replayId}`
+        );
 
         // Build list of channels
         let channels = [];
         EVENT_TYPES.forEach((eventType) => {
             const eventTypeName = eventType.value;
-            if (eventTypeName === EVT_CDC_STANDARD) {
-                // Use global channel for all CDC events
-                channels.push(CHANNEL_ALL_CDC);
-            } else {
-                // Get channels for specific event type
+            if (filter === FILTER_ALL) {
+                if (eventTypeName === EVT_CDC) {
+                    // Use global channel for all CDC events
+                    channels.push(CHANNEL_ALL_CDC);
+                } else {
+                    // Get all channels for the other event types
+                    const channelPrefix = getChannelPrefix(eventTypeName);
+                    this.channels[eventTypeName].forEach((channelData) => {
+                        channels.push(channelPrefix + channelData.value);
+                    });
+                }
+            } else if (filter === FILTER_CUSTOM) {
+                // Get custom channels for all event types
                 const channelPrefix = getChannelPrefix(eventTypeName);
                 this.channels[eventTypeName].forEach((channelData) => {
-                    channels.push(channelPrefix + channelData.value);
+                    if (isCustomChannel(eventTypeName, channelData.value)) {
+                        channels.push(channelPrefix + channelData.value);
+                    }
                 });
+            } else {
+                throw new Error(`Unsupported filter value: ${filter}`);
             }
         });
+
         // Remove already subscribed channels
         channels = channels.filter(
             (channel) =>
                 !this.subscriptions.some((sub) => sub.channel === channel)
         );
+
+        // Abort if there are no remaining channels
+        if (channels.length === 0) {
+            this.notify(
+                'warn',
+                'There are no channels to subscribe to with the specified filter and current subscriptions'
+            );
+            return;
+        }
+
+        // Temporarily ignore subscribe errors while subscribing to events
+        this.ignoreSubscribeErrors = true;
+        setTimeout(() => {
+            this.ignoreSubscribeErrors = false;
+        }, IGNORE_SUBCRIBE_ERRORS_DELAY);
+
         // Queue subscriptions
         const subscribePromises = channels.map((channel) => {
             return subscribe(channel, replayId, (streamingEvent) => {
                 this.handleStreamingEvent(streamingEvent);
             });
         });
+
         // Save susbcriptions and notify success once done
-        Promise.all(subscribePromises).then((subscriptions) => {
-            subscriptions.forEach((subscription) => {
-                this.saveSubscription(subscription);
-            });
-            this.notify('success', 'Successfully subscribed to all channels');
-            this.view = VIEW_MONITOR;
+        const subscriptions = await Promise.all(subscribePromises);
+        subscriptions.forEach((subscription) => {
+            this.saveSubscription(subscription);
         });
+        this.notify(
+            'success',
+            'Successfully subscribed to the specified channels'
+        );
+        this.view = VIEW_MONITOR;
     }
 
-    handleSubscribe(event) {
+    async handleSubscribe(event) {
         const { channel, replayId } = event.detail;
 
         // Check for duplicate subscription
@@ -196,17 +224,17 @@ export default class StreamingMonitor extends LightningElement {
             return;
         }
 
-        subscribe(channel, replayId, (streamingEvent) => {
-            this.handleStreamingEvent(streamingEvent);
-        }).then((subscription) => {
-            this.notify(
-                'success',
-                'Successfully subscribed',
-                subscription.channel
-            );
-            this.saveSubscription(subscription);
-            this.view = VIEW_MONITOR;
-        });
+        // Subscribe
+        const subscription = await subscribe(
+            channel,
+            replayId,
+            (streamingEvent) => {
+                this.handleStreamingEvent(streamingEvent);
+            }
+        );
+        this.notify('success', 'Successfully subscribed', subscription.channel);
+        this.saveSubscription(subscription);
+        this.view = VIEW_MONITOR;
     }
 
     saveSubscription(subscription) {
@@ -235,23 +263,19 @@ export default class StreamingMonitor extends LightningElement {
         this.eventsElement.addStreamingEvent(eventData);
     }
 
-    handlePublish(event) {
+    async handlePublish(event) {
         const eventParams = event.detail;
-        publishStreamingEvent(eventParams)
-            .then(() => {
-                this.notify(
-                    'success',
-                    `Successfully published event ${eventParams.eventName}`
-                );
-                this.view = VIEW_MONITOR;
-            })
-            .catch((error) => {
-                console.error(JSON.stringify(error));
-                this.notify(
-                    'error',
-                    `Failed to publish ${eventParams.eventName}`
-                );
-            });
+        try {
+            await publishStreamingEvent(eventParams);
+            this.notify(
+                'success',
+                `Successfully published event ${eventParams.eventName}`
+            );
+            this.view = VIEW_MONITOR;
+        } catch (error) {
+            console.error(JSON.stringify(error));
+            this.notify('error', `Failed to publish ${eventParams.eventName}`);
+        }
     }
 
     handleUnsubscribeAll() {