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.

The following golang metrics can be taken.

| name | type | desc | |---------------------|-----------|-----------------------------------------------------------------------| | connections | int64 | number of connected all clients | | connections_sse | int64 | number of connected SSE sclients | | connections_grpc | int64 | number of connected gRPC sclients |

Config

| name | type | desc