gremtune

README

gremtune

gremtune is a fork of qasaur/gremgo with alterations to make it compatible with AWS Neptune which is a "Fast, reliable graph database built for the cloud".

gremtune is a fast, efficient, and easy-to-use client for the TinkerPop graph database stack. It is a Gremlin language driver which uses WebSockets to interface with Gremlin Server and has a strong emphasis on concurrency and scalability. Please keep in mind that gremtune is still under heavy development and although effort is being made to fully cover gremtune with reliable tests, bugs may be present in several areas.

Modifications were made to gremgo in order to "support" AWS Neptune's lack of Gremlin-specific features, like no support query bindings among others. See differences in Gremlin support here: AWS Neptune Gremlin Implementation Differences

Installation

go get github.com/schwartzmx/gremtune
dep ensure

Documentation

• GoDoc

Example

package main

import (
    "fmt"
    "log"

    "github.com/schwartzmx/gremtune"
)

func main() {
    errs := make(chan error)
    go func(chan error) {
        err := <-errs
        log.Fatal("Lost connection to the database: " + err.Error())
    }(errs) // Example of connection error handling logic

    dialer := gremtune.NewDialer("ws://127.0.0.1:8182") // Returns a WebSocket dialer to connect to Gremlin Server
    g, err := gremtune.Dial(dialer, errs) // Returns a gremtune client to interact with
    if err != nil {
        fmt.Println(err)
        return
    }
    res, err := g.Execute( // Sends a query to Gremlin Server
        "g.V('1234')"
    )
    if err != nil {
        fmt.Println(err)
        return
    }
    j, err := json.Marshal(res[0].Result.Data) // res will return a list of resultsets,  where the data is a json.RawMessage
    if err != nil {
        fmt.Println(err)
        return nil, err
    }
    fmt.Printf("%s", j)
}

Example for streaming the result

Neptune provides 64 values per Response that is why Execute at present provides a [] of Response since it waits for all the responses to be retrieved and then provides it.In ExecuteAsync method it takes a channel to provide the Response as request parameter and provides the Response as and when it is provided by Neptune. The Response are streamed to the caller and once all the Responses are provided the channel is closed. go test -v -run ExecuteBulkDataAsync is the cmd to run the testcase)

package main

import (
    "fmt"
    "log"
    "time"
    "strings"
    "github.com/schwartzmx/gremtune"
)

func main() {
    errs := make(chan error)
    go func(chan error) {
        err := <-errs
        log.Fatal("Lost connection to the database: " + err.Error())
    }(errs) // Example of connection error handling logic

    dialer := gremtune.NewDialer("ws://127.0.0.1:8182") // Returns a WebSocket dialer to connect to Gremlin Server
    g, err := gremtune.Dial(dialer, errs) // Returns a gremtune client to interact with
    if err != nil {
        fmt.Println(err)
        return
    }
    start := time.Now()
    responseChannel := make(chan AsyncResponse, 10)
    err := g.ExecuteAsync( // Sends a query to Gremlin Server
        "g.V().hasLabel('Employee').valueMap(true)", responseChannel
    )
    log.Println(fmt.Sprintf("Time it took to execute query %s", time.Since(start)))
    if err != nil {
        fmt.Println(err)
        return
    }
    count := 0
    asyncResponse := AsyncResponse{}
    start = time.Now()
    for asyncResponse = range responseChannel {
        log.Println(fmt.Sprintf("Time it took to get async response: %s response status: %v", time.Since(start), asyncResponse.Response.Status.Code))
        count++
        
        nl := new(BulkResponse)
        datastr := strings.Replace(string(asyncResponse.Response.Result.Data), "@type", "type", -1)
        datastr = strings.Replace(datastr, "@value", "value", -1)
        err = json.Unmarshal([]byte(datastr), &nl)
        if err != nil {
           fmt.Println(err)
           return nil, err
        }
        log.Println(fmt.Sprintf("No of rows retrieved: %v", len(nl.Value)))
        start = time.Now()
    }
}

Authentication

The plugin accepts authentication creating a secure dialer where credentials are setted. If the server where are you trying to connect needs authentication and you do not provide the credentials the complement will panic.

package main

import (
    "fmt"
    "log"

    "github.com/schwartzmx/gremtune"
)

func main() {
    errs := make(chan error)
    go func(chan error) {
        err := <-errs
        log.Fatal("Lost connection to the database: " + err.Error())
    }(errs) // Example of connection error handling logic

    dialer := gremtune.NewSecureDialer("127.0.0.1:8182", "username", "password") // Returns a WebSocket dialer to connect to Gremlin Server
    g, err := gremtune.Dial(dialer, errs) // Returns a gremtune client to interact with
    if err != nil {
        fmt.Println(err)
        return
    }
    res, err := g.Execute( // Sends a query to Gremlin Server
        "g.V('1234')"
    )
    if err != nil {
        fmt.Println(err)
        return
    }
    j, err := json.Marshal(res[0].Result.Data) // res will return a list of resultsets,  where the data is a json.RawMessage
    if err != nil {
        fmt.Println(err)
        return nil, err
    }
    fmt.Printf("%s", j)
}

License

See LICENSE

Documentation

Index

• type AsyncResponse
• type Client
  • func Dial(conn dialer, errs chan error) (c Client, err error)
  • func (c *Client) Close()
  • func (c *Client) Execute(query string) (resp []Response, err error)
  • func (c *Client) ExecuteAsync(query string, responseChannel chan AsyncResponse) (err error)
  • func (c *Client) ExecuteFile(path string) (resp []Response, err error)
  • func (c *Client) ExecuteFileWithBindings(path string, bindings, rebindings map[string]string) (resp []Response, err error)
  • func (c *Client) ExecuteWithBindings(query string, bindings, rebindings map[string]string) (resp []Response, err error)
• type DialerConfig
  • func SetAuthentication(username string, password string) DialerConfig
  • func SetBufferSize(readBufferSize int, writeBufferSize int) DialerConfig
  • func SetPingInterval(seconds int) DialerConfig
  • func SetReadingWait(seconds int) DialerConfig
  • func SetTimeout(seconds int) DialerConfig
  • func SetWritingWait(seconds int) DialerConfig
• type Pool
  • func (p *Pool) Close()
  • func (p *Pool) Execute(query string) (resp []Response, err error)
  • func (p *Pool) ExecuteWithBindings(query string, bindings, rebindings map[string]string) (resp []Response, err error)
  • func (p *Pool) Get() (*PooledConnection, error)
• type PooledConnection
  • func (pc *PooledConnection) Close()
• type Response
  • func (r Response) ToString() string
• type Result
• type Status
• type Ws
  • func NewDialer(host string, configs ...DialerConfig) (dialer *Ws)
  • func (ws *Ws) IsConnected() bool
  • func (ws *Ws) IsDisposed() bool

Types

type AsyncResponse

type AsyncResponse struct {
	Response     Response `json:"response"`     //Partial Response object
	ErrorMessage string   `json:"errorMessage"` // Error message if there was an error
}

AsyncResponse structs holds the entire response from requests to the gremlin server

type Client

type Client struct {
	sync.RWMutex
	Errored bool
	// contains filtered or unexported fields
}

Client is a container for the gremtune client.

func Dial

func Dial(conn dialer, errs chan error) (c Client, err error)

Dial returns a gremtune client for interaction with the Gremlin Server specified in the host IP.

func (*Client) Close

func (c *Client) Close()

Close closes the underlying connection and marks the client as closed.

func (*Client) Execute

func (c *Client) Execute(query string) (resp []Response, err error)

Execute formats a raw Gremlin query, sends it to Gremlin Server, and returns the result.

func (*Client) ExecuteAsync

func (c *Client) ExecuteAsync(query string, responseChannel chan AsyncResponse) (err error)

Execute formats a raw Gremlin query, sends it to Gremlin Server, and the results are streamed to channel provided in method paramater.

func (*Client) ExecuteFile

func (c *Client) ExecuteFile(path string) (resp []Response, err error)

ExecuteFile takes a file path to a Gremlin script, sends it to Gremlin Server, and returns the result.

func (*Client) ExecuteFileWithBindings

func (c *Client) ExecuteFileWithBindings(path string, bindings, rebindings map[string]string) (resp []Response, err error)

ExecuteFileWithBindings takes a file path to a Gremlin script, sends it to Gremlin Server with bindings, and returns the result.

func (*Client) ExecuteWithBindings

func (c *Client) ExecuteWithBindings(query string, bindings, rebindings map[string]string) (resp []Response, err error)

ExecuteWithBindings formats a raw Gremlin query, sends it to Gremlin Server, and returns the result.

type DialerConfig

type DialerConfig func(*Ws)

DialerConfig is the struct for defining configuration for WebSocket dialer

func SetAuthentication

func SetAuthentication(username string, password string) DialerConfig

SetAuthentication sets on dialer credentials for authentication

func SetBufferSize

func SetBufferSize(readBufferSize int, writeBufferSize int) DialerConfig

SetBufferSize sets the read/write buffer size

func SetPingInterval

func SetPingInterval(seconds int) DialerConfig

SetPingInterval sets the interval of ping sending for know is connection is alive and in consequence the client is connected

func SetReadingWait

func SetReadingWait(seconds int) DialerConfig

SetReadingWait sets the time for waiting that reading occur

func SetTimeout

func SetTimeout(seconds int) DialerConfig

SetTimeout sets the dial timeout

func SetWritingWait

func SetWritingWait(seconds int) DialerConfig

SetWritingWait sets the time for waiting that writing occur

type Pool

type Pool struct {
	Dial        func() (*Client, error)
	MaxActive   int
	IdleTimeout time.Duration
	// contains filtered or unexported fields
}

Pool maintains a list of connections.

func (*Pool) Close

func (p *Pool) Close()

Close closes the pool.

func (*Pool) Execute

func (p *Pool) Execute(query string) (resp []Response, err error)

Execute grabs a connection from the pool, formats a raw Gremlin query, sends it to Gremlin Server, and returns the result.

func (*Pool) ExecuteWithBindings

func (p *Pool) ExecuteWithBindings(query string, bindings, rebindings map[string]string) (resp []Response, err error)

ExecuteWithBindings formats a raw Gremlin query, sends it to Gremlin Server, and returns the result.

func (*Pool) Get

func (p *Pool) Get() (*PooledConnection, error)

Get will return an available pooled connection. Either an idle connection or by dialing a new one if the pool does not currently have a maximum number of active connections.

type PooledConnection

type PooledConnection struct {
	Pool   *Pool
	Client *Client
}

PooledConnection represents a shared and reusable connection.

func (*PooledConnection) Close

func (pc *PooledConnection) Close()

Close signals that the caller is finished with the connection and should be returned to the pool for future use.

type Response

type Response struct {
	RequestID string `json:"requestId"`
	Status    Status `json:"status"`
	Result    Result `json:"result"`
}

Response structs holds the entire response from requests to the gremlin server

func (Response) ToString

func (r Response) ToString() string

ToString returns a string representation of the Response struct

type Result

type Result struct {
	// Query Response Data
	Data json.RawMessage        `json:"data"`
	Meta map[string]interface{} `json:"meta"`
}

Result struct is used to hold properties returned for results from requests to the gremlin server

type Status

type Status struct {
	Message    string                 `json:"message"`
	Code       int                    `json:"code"`
	Attributes map[string]interface{} `json:"attributes"`
}

Status struct is used to hold properties returned from requests to the gremlin server

type Ws

type Ws struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Ws is the dialer for a WebSocket connection

func NewDialer

func NewDialer(host string, configs ...DialerConfig) (dialer *Ws)

NewDialer returns a WebSocket dialer to use when connecting to Gremlin Server

func (*Ws) IsConnected

func (ws *Ws) IsConnected() bool

IsConnected returns whether the underlying websocket is connected

func (*Ws) IsDisposed

func (ws *Ws) IsDisposed() bool

IsDisposed returns whether the underlying websocket is disposed