memphis-functions.go

README

Cloud - Docs - X - YouTube

Memphis.dev is a highly scalable, painless, and effortless data streaming platform.
Made to enable developers and data teams to collaborate and build
real-time and streaming apps fast.

Installation

After installing and running memphis broker,
In your project's directory:

go get github.com/memphisdev/memphis-functions.go

Importing

import "github.com/memphisdev/memphis-functions.go/memphis"

Creating a Memphis function

Memphis provides a CreateFunction utility for more easily creating Memphis Functions.

The user will write a function which will act as an event handler and will be called for every event that is to be processed by Functions.

The user created event handler must fulfill the following function signature:

type EventHandlerFunction func([]byte, map[string]string, map[string]string) ([]byte, map[string]string, error)

The event handler will take in a []byte representation of an event, also a map[string]string of headers that belong to that event, and map[string]string representation of the function inputs.

The event handler will then return a modified version of these fields.

If the processing fails, the user function should return nil, nil, err. If the user wishes to skip over an event and not send it to the station, return nil, nil, nil and the event will be skipped. Events that return an error will be sent to the dead letter station.

package main

import (
	"encoding/json"
    "github.com/memphisdev/memphis-functions.go/memphis"
)

type Event struct {
	Field1 string `json:"field1"`
	Field2 string `json:"field2"`
}

func eventHandlerFunc(msgPayload[]byte, msgHeaders[string]string, inputs[string]string) ([]byte, map[string]string, error){
    // Get data from msgPayload
    var event Event
    json.Unmarshal(msgPayload, &event)
    
    // Modify or do something with the payload
    event.Field1 = "modified"
    
    // Return the payload back as []bytes
    eventBytes, _ := json.Marshal(event)
    return eventBytes, msgHeaders, nil
}

func main() {
	memphis.CreateFunction(eventHandlerFunc);
}

As mentioned previously, if the user would like to send the message to the dead letter station, simply return an error. The unproccessed payload and headers will be included with the message to the dead letter station.

package main

import (
	"encoding/json"
    "strings"
    "errors"
    "github.com/memphisdev/memphis-functions.go/memphis"
)

type Event struct {
	Field1 string `json:"field1"`
	Field2 string `json:"field2"`
}

func eventHandlerFunc(msgPayload[]byte, msgHeaders[string]string, inputs[string]string) ([]byte, map[string]string, error){
    // Get data from msgPayload
    var event Event
    json.Unmarshal(msgPayload, &event)
    
    // Modify or do something with the payload
    if strings.Contains(event.Field1, "Bob"){
        return nil, nil, errors.New("String had Bob in it!")
    } 
    
    // Return the payload back as []bytes
    eventBytes, _ := json.Marshal(event)
    return eventBytes, msgHeaders, nil
}

func main() {
	memphis.CreateFunction(eventHandlerFunc);
}

If the user would rather this message just be skipped, instead of being sent to the dead letter station, return nil for all values:

package main

import (
	"encoding/json"
    "strings"
    "github.com/memphisdev/memphis-functions.go/memphis"
)

type Event struct {
	Field1 string `json:"field1"`
	Field2 string `json:"field2"`
}

func eventHandlerFunc(msgPayload[]byte, msgHeaders[string]string, inputs[string]string) ([]byte, map[string]string, error){
    // Get data from msgPayload
    var event Event
    json.Unmarshal(msgPayload, &event)
    
    // Modify or do something with the payload
    if strings.Contains(event.Field1, "Bob"){
        return nil, nil, nil
    } 
    
    // Return the payload back as []bytes
    eventBytes, _ := json.Marshal(event)
    return eventBytes, msgHeaders, nil
}

func main() {
	memphis.CreateFunction(eventHandlerFunc);
}

Lastly, if the user is using a format like Protocol Buffers, the user may simple decode the msgPayload with their proto function. Assuming that we have a proto definition like:

syntax = "proto3";
package protobuf_example;

message Message{
    string data_field = 1;
}

That is saved in the directory user_message/message.pb.go, we can use this Message object like so:

package main

import (
	"encoding/json"
    "strings"
    "github.com/memphisdev/memphis-functions.go/memphis"
    "google.golang.org/protobuf/proto"
    "current_directory/user_message"
)

func eventHandlerFunc(msgPayload[]byte, msgHeaders[string]string, inputs[string]string) ([]byte, map[string]string, error){
    // Get data from msgPayload
    var my_message user_message.Message
    proto.Unmarshal(msgPayload, &user_message)
    
    // Modify or do something with the payload
    my_message.data_field = "new_data"
    
    // Return the payload back as []bytes
    eventBytes, _ := proto.Marshal(&my_message)
    return eventBytes, msgHeaders, nil
}

func main() {
	memphis.CreateFunction(eventHandlerFunc);
}