yuxh/gearman-go
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity

prepared for documentation

Browse Source
This commit is contained in:
Xing Xing 2013-12-25 17:01:42 +08:00
parent ddba5ab2f2
commit 02e6bfebcb
18 changed files with 76 additions and 288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
.hgtags
View File

@ -1,30 +0,0 @@
b68aee2a48811a1bee5994b56437c393c6fb2f5b 2011-05-24
0dc8bc71d7e895caf5803c1905bf07a823462fba native.start
dee83cac69e07ed4f3efde162d981f5855101845 0.0.1
dee83cac69e07ed4f3efde162d981f5855101845 go1-0.0.1
dee83cac69e07ed4f3efde162d981f5855101845 0.0.1
0000000000000000000000000000000000000000 0.0.1
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
b68aee2a48811a1bee5994b56437c393c6fb2f5b 2011-05-24
0000000000000000000000000000000000000000 2011-05-24
0dc8bc71d7e895caf5803c1905bf07a823462fba native.start
0000000000000000000000000000000000000000 native.start
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
a3b64f831c51ae215ce3b2c354483a4746e20555 go1-0.1
7928d7ed58bc0e36abebb5602b4f8880551054a7 go1-0.1
0000000000000000000000000000000000000000 0.0.1
dee83cac69e07ed4f3efde162d981f5855101845 0.0.1
7928d7ed58bc0e36abebb5602b4f8880551054a7 0.1
7928d7ed58bc0e36abebb5602b4f8880551054a7 go1-0.1
0000000000000000000000000000000000000000 go1-0.1
dee83cac69e07ed4f3efde162d981f5855101845 go1-0.0.1
0000000000000000000000000000000000000000 go1-0.0.1
4e3bf88517539cf6a0342e57b86df6e17ceb228c 0.1.1
4e3bf88517539cf6a0342e57b86df6e17ceb228c 0.1.1
0000000000000000000000000000000000000000 0.1.1
0000000000000000000000000000000000000000 0.1.1
eea0878b43d209630d7b342b1b99c61c839b454f 0.1.1
67f11fa2301f5e74a436883f66ce7fb121a4df82 0.1.2

40
README.md
View File

@ -3,8 +3,18 @@ Gearman-Go
[![Build Status](https://travis-ci.org/mikespook/gearman-go.png?branch=master)](https://travis-ci.org/mikespook/gearman-go)
This package is a [Gearman](http://gearman.org/) API for [Golang](http://golang.org).
It was implemented a native protocol for both worker and client API.
This module is a [Gearman](http://gearman.org/) API for the [Go Programming Language](http://golang.org).
The protocols were written in pure Go. It contains two sub-packages:
The client package is used for sending jobs to the Gearman job server,
and getting responses from the server.
"github.com/mikespook/gearman-go/client"
The worker package will help developers in developing Gearman worker
service easily.
"github.com/mikespook/gearman-go/worker"
Install
=======
@ -17,11 +27,10 @@ Install the worker package:
> $ go get github.com/mikespook/gearman-go/worker
Install both:
Both of them:
> $ go get github.com/mikespook/gearman-go
Usage
=====
@ -43,19 +52,22 @@ Usage
## Client
c, err := client.New("tcp4", "127.0.0.1:4730")
// ...
defer c.Close()
data := []byte("Hello\x00 world")
c.ErrHandler = func(e error) {
// ...
c, err := client.New("tcp4", "127.0.0.1:4730")
// ... error handling
defer c.Close()
c.ErrorHandler = func(e error) {
log.Println(e)
panic(e)
}
echo := []byte("Hello\x00 world")
echomsg, err := c.Echo(echo)
// ... error handling
log.Println(string(echomsg))
jobHandler := func(job *client.Job) {
log.Printf("%s", job.Data)
}
handle := c.Do("ToUpper", data, client.JOB_NORMAL, jobHandler)
// ...
handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
// ...
Branches
========
@ -64,9 +76,9 @@ Version 0.x means: _It is far far away from stable._
__Use at your own risk!__
* 0.1-testing Old API and some known issues, eg. [issue-14](https://github.com/mikespook/gearman-go/issues/14)
* 0.2-dev Refactoring a lot of things
* master current usable version
* 0.2-dev Refactoring a lot of things
* 0.1-testing Old API and some known issues, eg. [issue-14](https://github.com/mikespook/gearman-go/issues/14)
Authors
=======

9
client/client.go
View File

@ -1,15 +1,10 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (
"io"
"net"
"sync"
// "fmt"
// "fmt"
)
/*
@ -185,7 +180,7 @@ func (client *Client) handleResponse(key string, resp *Response) *Response {
}
// job handler
func (client *Client) handleInner(key string, resp *Response) * Response {
func (client *Client) handleInner(key string, resp *Response) *Response {
if h, ok := client.innerHandler[key]; ok {
h(resp)
delete(client.innerHandler, key)

5
client/common.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
const (

5
client/error.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (

5
client/pool.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (

5
client/request.go
View File

@ -1,8 +1,3 @@
// Copyright 2013 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (

5
client/response.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (

64
example/client/client.go
View File

@ -2,47 +2,47 @@ package main
import (
"github.com/mikespook/gearman-go/client"
"log"
"sync"
"log"
"sync"
)
func main() {
var wg sync.WaitGroup
// Set the autoinc id generator
var wg sync.WaitGroup
// Set the autoinc id generator
// You can write your own id generator
// by implementing IdGenerator interface.
// client.IdGen = client.NewAutoIncId()
// by implementing IdGenerator interface.
// client.IdGen = client.NewAutoIncId()
c, err := client.New("tcp4", "127.0.0.1:4730")
if err != nil {
log.Fatalln(err)
}
defer c.Close()
c.ErrorHandler = func(e error) {
log.Println(e)
}
echo := []byte("Hello\x00 world")
wg.Add(1)
echomsg, err := c.Echo(echo)
if err != nil {
log.Fatalln(err)
}
log.Println(string(echomsg))
wg.Done()
jobHandler := func(job *client.Job) {
log.Printf("%s", job.Data)
wg.Done()
}
handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
if err != nil {
log.Fatalln(err)
}
wg.Add(1)
defer c.Close()
c.ErrorHandler = func(e error) {
log.Println(e)
}
echo := []byte("Hello\x00 world")
wg.Add(1)
echomsg, err := c.Echo(echo)
if err != nil {
log.Fatalln(err)
}
log.Println(string(echomsg))
wg.Done()
jobHandler := func(job *client.Job) {
log.Printf("%s", job.Data)
wg.Done()
}
handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
if err != nil {
log.Fatalln(err)
}
wg.Add(1)
status, err := c.Status(handle)
if err != nil {
log.Fatalln(err)
}
log.Printf("%t", status)
if err != nil {
log.Fatalln(err)
}
log.Printf("%t", status)
wg.Wait()
wg.Wait()
}

BIN
example/worker/worker
View File

Binary file not shown.

15
gearman.go
View File

@ -3,10 +3,19 @@
// license that can be found in the LICENSE file.
/*
This module is Gearman API for golang.
The protocol was implemented by native way.
*/
This module is a Gearman API for the Go Programming Language.
The protocols were written in pure Go. It contains two sub-packages:
The client package is used for sending jobs to the Gearman job server,
and getting responses from the server.
import "github.com/mikespook/gearman-go/client"
The worker package will help developers in developing Gearman worker
service easily.
import "github.com/mikespook/gearman-go/worker"
*/
package gearman
import (

144
gearman_test.go
View File

@ -1,144 +0,0 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com> All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
/*
This module is Gearman API for golang.
The protocol was implemented by native way.
*/
package gearman
import (
"github.com/mikespook/gearman-go/client"
"github.com/mikespook/gearman-go/worker"
"strings"
"sync"
"testing"
"time"
)
const (
STR = "The gearman-go is a pure go implemented library."
GEARMAND = "127.0.0.1:4730"
)
func ToUpper(job worker.Job) ([]byte, error) {
data := job.Data()
data = []byte(strings.ToUpper(string(data)))
return data, nil
}
func Sleep(job worker.Job) ([]byte, error) {
time.Sleep(time.Second * 5)
return nil, nil
}
func TestJobs(t *testing.T) {
w := worker.New(worker.Unlimited)
if err := w.AddServer("tcp4", GEARMAND); err != nil {
t.Error(err)
return
}
defer w.Close()
t.Log("Servers added...")
if err := w.AddFunc("ToUpper", ToUpper, 0); err != nil {
t.Error(err)
return
}
if err := w.AddFunc("Sleep", Sleep, 0); err != nil {
t.Error(err)
return
}
t.Log("Functions added...")
w.ErrorHandler = func(e error) {
t.Error(e)
}
if err := w.Ready(); err != nil {
t.Error(err)
return
}
go w.Work()
t.Log("Worker is running...")
c, err := client.New("tcp4", GEARMAND)
if err != nil {
t.Error(err)
return
}
defer c.Close()
c.ErrorHandler = func(e error) {
t.Log(e)
}
{
var w sync.WaitGroup
jobHandler := func(job *client.Response) {
upper := strings.ToUpper(STR)
if string(job.Data) != upper {
t.Errorf("%s expected, got %s", upper, job.Data)
}
w.Done()
}
w.Add(1)
handle, err := c.Do("ToUpper", []byte(STR), client.JOB_NORMAL, jobHandler)
if err != nil {
t.Error(err)
return
}
w.Wait()
status, err := c.Status(handle)
if err != nil {
t.Error(err)
return
}
if status.Known {
t.Errorf("%s shouldn't be known", status.Handle)
return
}
if status.Running {
t.Errorf("%s shouldn't be running", status.Handle)
}
}
{
handle, err := c.DoBg("Sleep", nil, client.JOB_NORMAL)
if err != nil {
t.Error(err)
return
}
time.Sleep(time.Second)
status, err := c.Status(handle)
if err != nil {
t.Error(err)
return
}
if !status.Known {
t.Errorf("%s should be known", status.Handle)
return
}
if !status.Running {
t.Errorf("%s should be running", status.Handle)
}
}
{
status, err := c.Status("not exists handle")
if err != nil {
t.Error(err)
return
}
if status.Known {
t.Errorf("%s shouldn't be known", status.Handle)
return
}
if status.Running {
t.Errorf("%s shouldn't be running", status.Handle)
}
}
}

4
worker/agent.go
View File

@ -1,7 +1,3 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com> All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
import (

5
worker/common.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
const (

5
worker/error.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
import (

5
worker/inpack.go
View File

@ -1,8 +1,3 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
import (

6
worker/outpack.go
View File

@ -1,12 +1,6 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
import (
// "fmt"
"encoding/binary"
)

12
worker/worker.go
View File

@ -1,7 +1,3 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com> All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package worker
import (
@ -53,12 +49,12 @@ type Worker struct {
// Get a new worker
func New(limit int) (worker *Worker) {
worker = &Worker{
agents: make([]*agent, 0),
agents: make([]*agent, 0, limit),
funcs: make(JobFuncs),
in: make(chan *inPack, QUEUE_SIZE),
}
if limit != Unlimited {
worker.limit = make(chan bool, limit - 1)
worker.limit = make(chan bool, limit-1)
}
return
}
@ -161,7 +157,7 @@ func (worker *Worker) handleInPack(inpack *inPack) {
worker.err(err)
}
}()
if (worker.limit != nil) {
if worker.limit != nil {
worker.limit <- true
}
inpack.a.Grab()
@ -240,7 +236,7 @@ func (worker *Worker) SetId(id string) {
// Execute the job. And send back the result.
func (worker *Worker) exec(inpack *inPack) (err error) {
defer func() {
if (worker.limit != nil) {
if worker.limit != nil {
<-worker.limit
}
if r := recover(); r != nil {