# kafka-pipe

> A functional/fluent utility for kafka

Latest version **0.1.5-0** (published 2019-03-02) · ISC license · 0 weekly downloads

## Install

```sh
npm install kafka-pipe
pnpm add kafka-pipe
yarn add kafka-pipe
bun add kafka-pipe
```

## Health

**Score 15/100 (F)** — status: abandoned.

Positive: no vulnerabilities.

Warnings: low downloads; no types; no esm support; pre 1.0.

Negative: abandoned; low maintenance score.

## Facts

| | |
|---|---|
| Version | 0.1.5-0 |
| Published | 2019-03-02 |
| First published | 2019-02-27 |
| Weekly downloads | 0 |
| License | ISC |
| TypeScript types | none |
| Module format | CommonJS |
| Dependencies | 2 |
| Unpacked size | 323.1 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| Author | cj@specialblend.net |
| Maintainers | specialblend |
| Keywords | functional, kafka |

## Links

- npm: https://www.npmjs.com/package/kafka-pipe
- Repository: https://github.com/SpecialBlend/kafka-pipe
- Homepage: https://github.com/SpecialBlend/kafka-pipe#readme
- Issues: https://github.com/SpecialBlend/kafka-pipe/issues
- npm.io page: https://npm.io/package/kafka-pipe

## Dependencies (2)

- [ramda](https://npm.io/package/ramda.md) ^0.26.1
- [kafka-node](https://npm.io/package/kafka-node.md) ^4.0.2

## Recent versions

- 0.1.5-0 (latest) — 2019-03-02
- 0.1.4 — 2019-03-02
- 0.1.3 — 2019-02-28
- 0.1.2 — 2019-02-27
- 0.1.1 — 2019-02-27
- 0.1.0 — 2019-02-27

## README

# kafka-pipe

A functional/fluent utility for kafka, built on top of `kafka-node`.

## Features

### PipeConsumer

#### summary

class that extends kafka Consumer and adds a fluent `.pipe()` method

#### example

```
import { Client, PipeConsumer } from 'kafka-pipe'

const kafkaHost = 'example.com:9092'
const topic = 'test.topic'
const client = new Client({ kafkaHost })
const consumer = new PipeConsumer(client, [{ topic ])

consumer.pipe(console.log)

// prints 'hello, world!' to console.log
consumer.emit('message', 'hello, world!')

```

### createConsumer

#### summary

function that returns an instance of `PipeConsumer`

#### example

```
import * as R from 'ramda'
import { Client, createConsumer } from 'kafka-pipe'

const printMessage = ({ topic, value }) => {
    console.log('received message')
    console.log('topic: ', topic)
    console.log('message: ', value)
}

const sendMessageToS3 = R.pipe(
    R.prop('value'),
    R.reverse,
    R.tap(message => console.log('sending message to s3', message)),
    s3Put,
)

const kafkaHost = 'example.com:9092'

const topic = 'test.topic'

const client = new Client({ kafkaHost })

const consumer = createConsumer(client, topic)

consumer
    .pipe(printMessage)
    .pipe(sendMessageToS3)

```

### createSender

#### summary

higher order function that returns a curried `Producer.send` function which, when called, sends the given payload to the previously set topic

#### example

```
import { Client, createSender } from 'kafka-pipe'

const kafkaHost = 'example.com:9092'
const topic = 'test.topic'
const client = new Client({ kafkaHost })
const message = 'hello, world'
const sendToTestTopic = createSender(client, topic)

sendToTestTopic([message])

```

### createTransformer

#### summary

function that returns a `PipeConsumer`, which pipes messages from `sourceTopic`, thru provided `transform` function, then into `destinationTopic`

#### example

```
import { Client, createTransformer } from 'kafka-pipe'

const normalizeMessage = () => {
    // normalize message before forwarding ...
}

const kafkaHost = 'example.com:9092'
const sourceTopic = 'test.source.topic'
const destinationTopic = 'test.destination.topic'
const client = new Client({ kafkaHost })
const consumer = createTransformer(client, sourceTopic, destinationTopic, normalizeMessage) 

```

---
_Source: https://npm.io/package/kafka-pipe · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._