JarCastingpbs
Streaming encoder/decoder for protocol buffers
npm install pbs
Usage
var pbs = require('pbs')
var messages = pbs(`
message Company {
required string name = 1;
repeated Employee employees = 2;
optional string country = 3;
message Employee {
required string name = 1;
required uint32 age = 2;
}
}
`)
// create a streaming encoder
var encoder = messages.Company.encode()
// create a streaming decoder
var decoder = messages.Company.decode()
Encoding
Use pbs to encode a protocol buffers message (no matter how large!) to a stream.
The encoder stream will expose all properties of the protobuf message as methods on the stream that you can pass the value you want to write to.
encoder.someProperty(aValue, [callback])
The callback is called when the stream has been flushed.
Here is an example using the above protobuf schema:
// all the properties of Company are exposed as methods
var encoder = messages.Company.encode()
// encoder is a readable stream containing the protobuf message.
// you can pipe it anywhere!
encoder.pipe(fs.createWriteStream('my-protobuf-message.pb'))
// write a name to the stream
encoder.name('my-company')
// write an employee to the stream
encoder.employees({
name: 'mathias',
age: 28
})
// write another one
encoder.employees({
name: 'jane doe',
age: 32
})
// no more data - will end the readable stream
encoder.finalize()
The encoder stream produces a valid protobuf message that can be decoded with any other parser that follows the protobuf spec.
Decoding
Similar to encoding you can use pbs to decode a protobuf message.
The decoder stream also exposes the properties as methods, but instead of passing a value you pass a function that is called then that property is found in the stream.
decoder.someProperty(fn)
Here is an example using the above schema:
// all the properties of Company are exposes as methods
var decoder = messages.Company.decode()
decoder.name(function (name, cb) {
console.log('message has name:', name)
cb() // done processing
})
decoder.employees(function (employee, cb) {
console.log('employee array member:', employee)
cb() // done processing
})
decoder.country(function (country, cb) {
console.log('message has country:', country)
cb()
})
decoder.on('finish', function () {
console.log('(no more data)')
})
fs.createReadStream('my-protobuf-message.pb').pipe(decoder)
Use cases
You can use this to parse large protobuf messages that might not fit in memory.
Another use case is to use this to implement a streaming binary protocol encoder/decoder. Let's say I wanted to implement a chat protocol. I could describe it using the following proto schema:
message ChatProtocol {
repeated Message messages = 1;
repeated string online = 2;
message Message {
required string from = 1;
required string text = 2;
}
}
and then just use pbs to parse it:
var fs = require('fs')
var pbs = require('pbs')
var messages = pbs(fs.readFileSync('schema.proto'))
var decoder = messages.ChatProtocol.decode()
// read messages
decoder.online(function (username, cb) {
console.log(username + ' is online!')
cb()
})
decoder.messages(function (message, cb) {
console.log(message.from + ' says: ' + message.text)
})
// write messages
var encoder = messages.ChatProtocol.encode()
encoder.online('mafintosh')
encoder.messages({
from: 'mafintosh',
text: 'hello world!'
})
// setup the pipeline
encoder.pipe(someTransportStream).pipe(decoder)
Since the entire stream is valid protobuf, you could even save it to a file and parse it using another protobuf parser to debug an application.
License
MIT