app.js

@@ -13,31 +13,23 @@ const { buildEscape } = require('./safely-exit')
 const config = readConfig()
 console.log('CONFIG', config)
 
-/**
+const c = object => {
- * Log out an object and return it.
+  console.log(object)
- *
+  return object
- * @param object The object to log
- * @param f Optional param to specify an alternative log function. E.g. console.error
- * @returns {*}
- */
-const c = (object, f = console.log) => {
-  f(object)
-  return object;
 }
 
 const buildCluster = (clusterConfig, connect = true) => {
   const kafkaConfig = {...clusterConfig}
-  delete kafkaConfig.clusterName // new Kafka() tries to use this same value
+  delete kafkaConfig.clusterName
 
   const kafka = new Kafka(kafkaConfig)
   const admin = kafka.admin()
-  const cluster = {
+  admin.connect().catch(console.error)
+  return c({
     kafka,
     admin,
     config: clusterConfig
-  }
+  })
-  admin.connect().catch(e => console.error(cluster.error = e.toString()))
-  return cluster
 }
 
 const clusters =
@@ -48,7 +40,7 @@ console.log('CLUSTERS', clusters)
 const app = express();
 
 app.use(cors({
-  origin: config.frontendUrl,
+  origin: 'http://localhost:5173'
 }))
 app.use(logger('dev'));
 app.use(express.json());
@@ -58,45 +50,29 @@ app.use(express.static(path.join(__dirname, 'public')));
 
 const router = express.Router()
 
-process.on('unhandledRejection', console.error)
-
 /* GET topics listing. */
 router.get('/topics/:cluster', async (req, res, _next) => {
   const legalName = topicName => !topicName.startsWith("__")
-  try {
+  const topicList = await clusters[req.params.cluster]?.admin.listTopics()
-    const topicList = (await clusters[req.params.cluster]?.admin.listTopics() || [])
+  if (topicList) {
     res.send(topicList.filter(legalName))
-  } catch (e) {
+  } else {
-    res.status(502).send({
+    res.send([])
-      error: `Could not connect to cluster '${req.params.cluster}'`,
+    //res.status(400).send({error: 'Invalid cluster name'})
-      errorDetails: e.toString()
-    })
   }
 })
 
-const buildSocketMessageSender = ({ type, socket }) => message => {
-  socket.send(JSON.stringify({
-    type,
-    message
-  }))
-}
-
-const deepCopy = object => JSON.parse(JSON.stringify(object))
-
 const passwordPlaceholder = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXX'
 
 const getClusterData = () =>
   Object.fromEntries(
-    Object.entries(clusters).map(([key, cluster]) => {
+    Object.entries(clusters).map(([key, value]) => {
-      cluster = deepCopy(cluster)
+      value = JSON.parse(JSON.stringify(value))
-      if (cluster.config.sasl?.password) {
+      value.config.sasl.password = passwordPlaceholder
-        cluster.config.sasl.password = passwordPlaceholder
+      return [key, value.config]
-      }
-      return [key, cluster.config]
     }))
 
 router.get('/clusters', async (req, res, _next) => {
-  console.log('/clusters')
   res.send(getClusterData())
 })
 
@@ -110,17 +86,9 @@ router.post('/clusters', async (req, res, _next) => {
 })
 
 router.post('/clusters/delete', async (req, res, _next) => {
+  console.log('/clusters/delete post body', req.body)
   const clusterName = req.body.clusterName
-  // Kill all consumers connected to this cluster and notify those clients via their websocket
+  // TODO: Disconnect
-  consumers.forEach(([{ consumer, cluster }, socket]) => {
-    if (clusters[clusterName] !== cluster) {
-      // TODO This reference equality may not be sufficient?
-      return
-    }
-    killConsumer(consumer)
-    const sendDeletedMessage = buildSocketMessageSender({ type: 'cluster_deleted', socket })
-    sendDeletedMessage(clusterName)
-  })
   delete clusters[clusterName]
   delete config.clusters[clusterName]
   await storeConfig(config)
@@ -128,7 +96,10 @@ router.post('/clusters/delete', async (req, res, _next) => {
 })
 
 router.put('/clusters', async (req, res, _next) => {
+  console.log('/clusters put body', req.body)
+
   const hasPlaceholderPassword = req.body.sasl.password === passwordPlaceholder
+  console.log('Has placeholder password:', hasPlaceholderPassword)
   const clusterName = req.body.clusterName
   if (hasPlaceholderPassword) {
     req.body.password = config.clusters[clusterName].password
@@ -136,72 +107,83 @@ router.put('/clusters', async (req, res, _next) => {
   config.clusters[clusterName] = req.body
   clusters[clusterName] = buildCluster(req.body)
   res.send('')
+  //res.send(getClusterData())
   await storeConfig(config)
 })
 
 app.use(router)
 
-const realTimeSearch = async ({ kafka, socket, topic }) =>
+const realTimeSearch = async ({ kafka, searchCode, immutable, socket, topic }) =>
   query.realTimeMessageSearch({
     kafka,
     topic,
-    onMessage: buildSocketMessageSender({ socket, type: 'message' })
+    searchCode,
-  });
+    immutable,
+    onMessage: message => {
+      socket.send(JSON.stringify({
+        type: 'message',
+        message
+      }))
+    }
+  })
 
-const oneShotSearch = async ({ kafka, socket, topic, maxItems }) =>
+const search = async ({ kafka, searchCode, immutable, socket, topic, maxItems }) =>
   query.searchMessages({
     kafka,
     topic,
     maxItems,
-    onDone: buildSocketMessageSender({ socket, type: 'count' }),
+    searchCode,
-    onBatchDone: buildSocketMessageSender({ socket, type: 'item_count' })
+    immutable,
+    onDone: messages => {
+      console.log(messages.length + ' messages')
+      socket.send(JSON.stringify({
+        type: 'complete',
+        message: messages
+      }))
+    },
+    onBatchDone: count => {
+      socket.send(JSON.stringify({
+        type: 'item_count',
+        message: count
+      }))
+    }
   })
 
-/** @type {WebSocketServer} */
 const wsServer = new ws.WebSocketServer({
   noServer: true
 })
 
-/** @type {Map<WebSocket, ({consumer: kafka.Consumer, cluster: { kafka, admin, config }})>} */
 const consumers = new Map()
 
 buildEscape(consumers, clusters)
 
 wsServer.on('connection', socket => {
-  socket.send('CONNECTED')
   socket.on('close', async () => {
-    await killConsumer(consumers.get(socket).consumer)
+    await killConsumer(consumers.get(socket))
     consumers.delete(socket)
   })
 
   socket.on('message', async message => {
+    socket.send('Loading...')
     message = JSON.parse(message)
 
-    if (message.mode === 'kill') {
+    const currentMode = message.mode === 'realTime' ? realTimeSearch : search
-      console.log('KILLING SOCKET')
+    console.log('CLUSTERS before run', clusters)
-      await killConsumer(consumers.get(socket).consumer)
+    console.log('message.cluster', message.cluster)
-      consumers.delete(socket)
-      return
-    }
-    const startSearch = message.mode === 'realTime' ? realTimeSearch : oneShotSearch
     const cluster = clusters[message.cluster]
-
+    console.log('clusters[message.cluster]', cluster)
-    const run = async () => {
+    const run = async () => consumers.set(socket, await currentMode({
-      const consumerCluster = {
-        consumer: await startSearch({
       kafka: cluster.kafka,
+      searchCode: message.searchCode,
+      immutable: message.immutable,
       topic: message.topic,
       maxItems: message.maxItems,
       socket
-        }),
+    }))
-        cluster
-      }
-      consumers.set(socket, consumerCluster);
-    }
 
     run().catch(async e => {
       console.error('run() error occurred!', e.toString())
-      await killConsumer(consumers.get(socket).consumer)
+      await killConsumer(consumers.get(socket))
       // Try again ONCE on failure
       run().catch(ee => socket.send('ERROR: ' + ee))
     })

config.js

@@ -2,22 +2,22 @@ const fs = require('fs')
 const path = require('path')
 const homedir = require('os').homedir()
 
-const configFilePath = homedir + path.sep + '.kafka-dance.json'
+const configFilePath = homedir + path.sep + '.kafka-dance'
 
 const emptyConfig = () => ({
-  clusters: {},
+  clusters: {}
-  frontendUrl: 'http://localhost:5173',
 })
 
 const readConfig = () => {
-  if (fs.existsSync(configFilePath)) {
+  if (!fs.existsSync(configFilePath)) {
+    return emptyConfig()
+  }
   try {
     return JSON.parse(fs.readFileSync(configFilePath).toString())
   } catch (e) {
     console.error(e.toString())
-    }
-  }
     return emptyConfig()
+  }
 }
 
 const storeConfig = async config =>

query.js

@@ -1,55 +1,60 @@
-const parseValue = message => {
+const buildMatcher = code => new Function('message', 'value', code)
-  const s = message.value.toString()
-  try {
-    return JSON.parse(s)
-  } catch (e) {
-    console.error('Error deserializing message.value', e) // But keep the value as a string
-    return '' + s
-  }
-}
 
-const buildConsumer = async ({ kafka, topic, fromBeginning }) => {
+const deepCopy = obj => JSON.parse(JSON.stringify(obj))
-  // TODO Use a UUID/GUID instead
-  const consumer = kafka.consumer({ groupId: 'sage-testing-group' + Math.random() })
 
+const realTimeMessageSearch = async ({ kafka, topic, onMessage, searchCode, immutable = true }) => {
+  const consumer = kafka.consumer({ groupId: 'sage-testing-group' })
+
+  console.log({ topic, onMessage, searchCode, immutable })
   await consumer.connect()
-  await consumer.subscribe({ topic, fromBeginning })
+  await consumer.subscribe({ topic, fromBeginning: true })
+  const matcher = searchCode && buildMatcher(searchCode)
 
-  return consumer
+  const messages = []
-}
-
-const seekToBeginning = async ({ topic, consumer, partitionCount }) => {
-  for (let i = 0; i <= partitionCount; i++) {
-    await consumer.seek({topic, partition: i, offset: 0})
-  }
-}
-
-const realTimeMessageSearch = async ({ kafka, topic, onMessage }) => {
-  // TODO Use a UUID/GUID instead
-  const consumer = await buildConsumer({ kafka, topic, fromBeginning: true })
 
+  console.log('consumer.run')
   await consumer.run({
     autoCommit: false,
     eachMessage: async ({ topic, partition, message }) => {
-      message.value = parseValue(message.value)
+      const s = message.value.toString()
+      const value = JSON.parse(s)
+      delete value.Data
+      message.value = value
+      const workingMessage = immutable ? deepCopy(message) : message
+      if (!matcher || matcher(workingMessage, workingMessage.value)) {
         onMessage(message)
+        messages.push(message)
+        //console.log(message)
+      }
     },
   })
 
-  // seekToBeginning({ topic, consumer, partitionCount: 5 })
+  for (let i = 0; i <= 5; i++) {
-
+    //console.log(`seek on partition ${i}`)
+    await consumer.seek({topic, partition: i, offset: 0})
+  }
+  console.log('return consumer')
   return consumer
 }
 
-const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) => {
+const searchMessages = async ({ kafka, maxItems, topic, onDone, searchCode, onBatchDone, immutable = true }) => {
-  const consumer = await buildConsumer({ kafka, topic, fromBeginning: true })
+  const consumer = kafka.consumer({ groupId: 'sage-testing-group' })
+
+  await consumer.connect()
+  await consumer.subscribe({ topic, fromBeginning: true })
+  const matcher = searchCode && buildMatcher(searchCode)
 
   const messages = []
 
   await consumer.run({
     autoCommit: false,
     eachMessage: async ({ topic, partition, message }) => {
-      message.value = parseValue(message.value)
+      const s = message.value.toString()
+      const value = JSON.parse(s)
+      delete value.Data
+      message.value = value
+      let workingMessage = immutable ? deepCopy(message) : message
+      if (!matcher || matcher(workingMessage, workingMessage.value)) {
         messages.push(message)
         if (messages.length < maxItems) {
           onBatchDone(messages.length)
@@ -59,6 +64,8 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
           await consumer.disconnect()
           onDone(messages.slice(0, maxItems))
         }
+        //console.log(message)
+      }
     },
   })
 
@@ -81,6 +88,7 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
    */
   let consumedTopicPartitions = {}
   consumer.on(consumer.events.GROUP_JOIN, async ({ payload }) => {
+    //console.log('GROUP_JOIN')
     const { memberAssignment } = payload
     consumedTopicPartitions = Object.entries(memberAssignment).reduce(
       (topics, [topic, partitions]) => {
@@ -93,7 +101,6 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
     )
   })
 
-  // This was part of the original code, but seems to cause a near-immediate exit
   /*
    * This is extremely unergonomic, but if we are currently caught up to the head
    * of all topic-partitions, we won't actually get any batches, which means we'll
@@ -107,6 +114,7 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
   //let processedBatch = true
   consumer.on(consumer.events.FETCH_START, async () => {
     //console.log('FETCH_START')
+    // This was part of the original code, but seems to cause a near-immediate exit
     //if (processedBatch === false) {
     //  console.log('I DIE AT FETCH')
     //  await consumer.disconnect()
@@ -121,6 +129,7 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
    * and exit if all topic-partitions have been consumed,
    */
   consumer.on(consumer.events.END_BATCH_PROCESS, async ({ payload }) => {
+    //console.log('END_BATCH_PROCESS')
     const { topic, partition, offsetLag } = payload
     consumedTopicPartitions[`${topic}-${partition}`] = offsetLag === '0'
 
@@ -138,6 +147,10 @@ const searchMessages = async ({ kafka, maxItems, topic, onDone, onBatchDone }) =
   return consumer
 }
 
+//run()
+//  .then(() => console.log('DONE'))
+//  .catch(e => console.error(`[example/consumer] ${e.message}`, e))
+
 module.exports = {
   searchMessages,
   realTimeMessageSearch,

safely-exit.js

@@ -4,7 +4,7 @@ const signalTraps = ['SIGTERM', 'SIGINT', 'SIGUSR2']
 const disconnectAll = async (consumers, clusters) => {
   await Promise.all([
     ...Object.values(clusters).map(async cluster => cluster.admin.disconnect()),
-    ...Array.from(consumers.values()).map(async consumer => consumer.consumer.disconnect())
+    ...Array.from(consumers.values()).map(async consumer => consumer.disconnect())
   ])
 }