bucket notifications - kafka topic notation

Signed-off-by: Amit Prinz Setter <[email protected]>

Author: alphaprinz

src/test/unit_tests/test_notifications.js

@@ -55,7 +55,6 @@ let expected_event_name;
 let expected_key;
 let expected_eTag;
 let expect_test;
-let expected_url;
 
 // eslint-disable-next-line max-lines-per-function
 mocha.describe('notifications', function() {
@@ -96,7 +95,6 @@ mocha.describe('notifications', function() {
                     assert.strictEqual(notif.Records[0].Event, "s3:TestEvent", 'wrong event name in notification');
                     expect_test = false;
                 } else {
-                    assert.strictEqual(req.url, expected_url);
                     assert.strictEqual(notif.Records[0].s3.bucket.name, expected_bucket, 'wrong bucket name in notification');
                     assert.strictEqual(notif.Records[0].eventName, expected_event_name, 'wrong event name in notification');
                     assert.strictEqual(notif.Records[0].s3.object.key, expected_key, 'wrong key in notification');
@@ -237,34 +235,6 @@ mocha.describe('notifications', function() {
             });
         });
 
-        mocha.it('override connection', async () => {
-            await s3.putBucketNotificationConfiguration({
-                Bucket: bucket,
-                NotificationConfiguration: {
-                    TopicConfigurations: [{
-                        "Id": "system_test_http_no_event_override",
-                        "TopicArn": http_connect_filename + "?" + JSON.stringify({
-                            request_options_object: {path: "/override"}
-                        }),
-                    }],
-                },
-            });
-
-            const res = await s3.putObject({
-                Bucket: bucket,
-                Key: 'f1',
-                Body: 'this is the body',
-            });
-
-            await notify_await_result({
-                bucket_name: bucket,
-                event_name: 'ObjectCreated:Put',
-                key: "f1",
-                etag: res.ETag,
-                url: "/override"
-            });
-        });
-
     });
 
 });
@@ -277,7 +247,6 @@ async function notify_await_result({bucket_name, event_name, etag, key, url = "/
     expected_event_name = event_name;
     expected_eTag = etag;
     expected_key = key;
-    expected_url = url;
     server_done = false;
 
     //busy-sync wait for server

src/util/notifications_util.js

@@ -186,24 +186,23 @@ class Notificator {
         }
     }
 
-    async parse_connect_file(connect_filename_with_overrides, decrypt = false) {
+    async parse_connect_file(connection_name, decrypt = false) {
         let connect;
-        const filename_parts = connect_filename_with_overrides.split('?');
-        const connect_filename_no_overrides = filename_parts[0];
-        const overrides_str = filename_parts[1];
+        let connect_filename = connection_name;
+        let kafka_topic_from_connection_name;
+        if (connection_name.startsWith("kafka:::topic/")) {
+            const connection_parts = connection_name.split('/');
+            connect_filename = connection_parts[1];
+            kafka_topic_from_connection_name = connection_parts.length > 2 && connection_parts[3];
+        }
 
         if (this.nc_config_fs) {
-            connect = await this.nc_config_fs.get_connection_by_name(connect_filename_no_overrides);
+            connect = await this.nc_config_fs.get_connection_by_name(connect_filename);
         } else {
-            const filepath = path.join(this.connect_files_dir, connect_filename_no_overrides);
+            const filepath = path.join(this.connect_files_dir, connect_filename);
             const connect_str = fs.readFileSync(filepath, 'utf-8');
             connect = JSON.parse(connect_str);
         }
-        if (overrides_str) {
-            const overrides_obj = JSON.parse(overrides_str);
-            _.merge(connect, overrides_obj);
-            dbg.log2("effective connect =", connect);
-        }
 
         //if connect file is encrypted (and decryption is requested),
         //decrypt the auth field
@@ -212,6 +211,11 @@ class Notificator {
                 connect.request_options_object.auth, connect.master_key_id);
         }
         load_files(connect);
+
+        //use the kafka topic, if it was present in connection_name
+        if (kafka_topic_from_connection_name) {
+            connect.topic = kafka_topic_from_connection_name;
+        }
         return connect;
     }
 }
@@ -371,6 +375,7 @@ async function test_notifications(notifs, nc_config_dir, req) {
         let notif_failure;
         try {
             connect = await notificator.parse_connect_file(notif.topic[0]);
+            dbg.log0(`effective connect for notif ${notif.id[0]} is`, connect);
             connection = get_connection(connect);
             await connection.connect();
             await connection.promise_notify(compose_notification_test(req), async (notif_cb, err_cb, err) => {