plasma

Attention: This repository was transfererd from openfresh.

plasma is event push middleware by using gRPC stream.

Description

Plasma is middleware for sending event specialized for a stream. Plasma provides EventSource and gRPC Stream from the same endpoint.

Installation

This middleware requires Redis.

From Source

$ git clone git://github.com/openfresh/plasma.git $GOPATH/src/github.com/openfresh/plasma
$ cd  $GOPATH/src/github.com/openfresh/plasma
$ make deps
$ make build

The binary is generated under the bin/ directory.

Using docker

You can also use the Docker image.

$ docker run -p 8080:8080 -p 50051:50051 -p 9999:9999 openfresh/plasma

Using docker-compose

You can use docker-compose for easy use without preparing Redis.

$ git clone git://github.com/openfresh/plasma.git $GOPATH/src/github.com/openfresh/plasma
$ cd  $GOPATH/src/github.com/openfresh/plasma
$ docker-compose up -d

Usage Subscriber

Server Sent Events

Using server-sent events

Your can use SSE if you request with Accept:text-stream header.

You request events that you want to subscribe to this endpoint. You can specify multiple events separated by commas. The query name can be set with the EventQuery environment variable．(default value of EventQuery is eventType )．

Here is a simple example using [Yaffle / EventSource] (https://github.com/Yaffle/EventSource).

    var source = new EventSource('//localhost:8080/?eventType=program:1234:views,program:1234:poll,program:1234:annotation');
    
    source.addEventListener("open", function(e) {
        console.log("open");
    });
    
    source.addEventListener("error", function(e) {
        console.log("error");
    });
    
    source.addEventListener("message", function(e) {
        console.log("message event: ", e.data);
    });

The JSON schema of data returned from Plasma is as follows.

{
    "$schema": "http://json-schema.org/draft-04/schema#",
    "id": "/events",
    "properties": {
        "data": {
            "id": "/events/data",
            "type": "string"
        },
        "meta": {
            "id": "/events/meta",
            "properties": {
                "type": {
                    "id": "/events/meta/type",
                    "type": "string"
                }
            },
            "type": "object"
        }
    },
    "type": "object"
}

If the DEBUG environment variable is enabled, you can access the debug endpoint.

GET /debug

You can publish events to Redis from this endpoint. You need to enter valid JSON in EventData form.

gRPC Stream

You can subscribe to events using gRPC Stream.

The ProtocolBuffer file is here .

The following is a simple Go sample.

func main() {
    conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()

    client := proto.NewStreamServiceClient(conn)
    ctx := context.Background()

    req := proto.Request{
        Events: []*proto.EventType{
            eventType("program:1234:poll"),
            eventType("program:1234:views"),
        },
    }

    ss, err := client.Events(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // subscribe event
    if err := ss.Send(&req); err != nil {
        log.Fatal(err)
    }

    for {
        resp, err := ss.Recv()
        if err != nil {
            log.Println(err)
            continue
        }
        if resp == nil {
            log.Println("payload is nil")
            continue
        }
        fmt.Printf("Meta: %s\tData: %s\n", resp.EventType.Type, resp.Data)
    }
}

unsubscribe

Events request is stream. If you unsubscribe event, set empty event data.

    req := proto.Request{
        // empty events
        Events: []*proto.EventType{},
    }

    ss, err := client.Events(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // unsubscribe event
    if err := ss.Send(&req); err != nil {
        log.Fatal(err)
    }

Usage Publisher

You publish events to the channel that Plasma subscribes according to the following JSON Schema.

{
    "$schema": "http://json-schema.org/draft-04/schema#",
    "id": "/events",
    "properties": {
        "data": {
            "id": "/events/data",
            "type": "string"
        },
        "meta": {
            "id": "/events/meta",
            "properties": {
                "type": {
                    "id": "/events/meta/type",
                    "type": "string"
                }
            },
            "type": "object"
        }
    },
    "type": "object"
}

openfresh/plasma-go is a library that wraps publish an event to Redis.

HealthCheck

GET /hc

You can do a health check. Check the status of Redis. If there is a problem it returns 500, and if there is no problem it returns 200.

Metrics

GET /metrics/go

You can get golang metrics from this endpoint.

The following golang metrics can be taken.

name	type	desc
go_version	string	version of go
go_os	string	os
go_arch	string	arch
cpu_num	int	number of cpus
goroutine_num	int	number of goroutines
gomaxprocs	int	number of operating system threads
cgo_call_num	int64	number of cgo calls
memory_alloc	uint64	bytes of allocated heap objects
memory_total_alloc	uint64	cumulative bytes allocated for heap objects
memory_sys	uint64	total bytes of memory
memory_lookups	uint64	number of pointer
memory_mallocs	uint64	cumulative count of heap objects allocated
memory_frees	uint64	cumulative count of heap objects freed
memory_stack	uint64	bytes in stack spans
heap_alloc	uint64	bytes of allocated heap objects
heap_sys	uint64	bytes of heap memory obtained
heap_idle	uint64	bytes in idle spans
heap_inuse	uint64	bytes in in-use spans
heap_released	uint64	bytes of physical memory returned to the OS
heap_objects	uint64	number of allocated heap objects
gc_next	uint64	target heap size of the next GC cycle
gc_last	uint64	time the last garbage collection finished, as, nanoseconds since 1970
gc_num	uint32	number of completed GC cycles
gc_per_second	float64	GC per second
gc_pause_per_second	float64	GC pause per second
gc_pause	[]float64	GC pause

GET /metrics/plasma

You can get plasma metrics from this endpoint.