59 lines
1.3 KiB
Go
59 lines
1.3 KiB
Go
package lib
|
|
|
|
import (
|
|
producer_module "com.navi.medici.janus/producer"
|
|
|
|
// "log"
|
|
"net/http"
|
|
"encoding/binary"
|
|
"github.com/Shopify/sarama"
|
|
|
|
)
|
|
|
|
var (
|
|
RequestChannel = make(chan *RequestObject)
|
|
)
|
|
|
|
type RequestObject struct {
|
|
Body []byte
|
|
Header http.Header
|
|
}
|
|
|
|
func ProcessRequestChannel(topic string) {
|
|
for {
|
|
request := <- RequestChannel
|
|
ClickstreamProtobufEventHandler(*request, topic)
|
|
}
|
|
}
|
|
|
|
func ClickstreamProtobufEventHandler(request RequestObject, topic string) {
|
|
|
|
messageBytes := request.Body
|
|
|
|
// to be of the format [magicByte] + [schemaID] + [messageIndex] + [value]
|
|
recordValue := []byte{}
|
|
|
|
// add [magicByte]
|
|
recordValue = append(recordValue, byte(0))
|
|
|
|
// add schemaID]
|
|
schemaIDBytes := make([]byte, 4)
|
|
binary.BigEndian.PutUint32(schemaIDBytes, uint32(producer_module.SchemaVersionMap[topic]))
|
|
recordValue = append(recordValue, schemaIDBytes...)
|
|
|
|
// add [messageIndex]
|
|
messageIndexBytes := []byte{byte(2), byte(0)}
|
|
recordValue = append(recordValue, messageIndexBytes...)
|
|
|
|
// Now write the bytes from the actual message...
|
|
recordValue = append(recordValue, messageBytes...)
|
|
|
|
message := &sarama.ProducerMessage{
|
|
Topic: topic,
|
|
Value: sarama.ByteEncoder(recordValue),
|
|
}
|
|
|
|
producer_module.WriteMessageToKafkaAsync(message)
|
|
|
|
}
|