init

LICENSE

@@ -0,0 +1,23 @@
+The MIT License (MIT)
+
+
+opyright (c) <2015> <Damnever([email protected])>
+
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README.md

@@ -0,0 +1,15 @@
+## A simple thread safe queue for Golang
+
+Golang channel can not support infinite size, I use it as a replacement.
+
+Queue Methods(maybe more):
+
+ - `NewQueue() *Queue`
+ - `IsEmpty() bool`
+ - `Size() int`
+ - `Get(block bool, timeout float64) (*list.Element, error)`
+ - `Put(value interface) *list.Element`
+
+---
+
+NOTE: Maybe a lot of BUGs in the code...

makefile

@@ -0,0 +1,3 @@
+test:
+	export GOPATH=$(pwd)
+	go test

queue.go

@@ -0,0 +1,100 @@
+/*
+  A thread safe queue written in Go.
+  Golang do not support infinite size channel.
+  It can get element block with timeout.
+*/
+package goqueue
+
+import (
+	"container/list"
+	"errors"
+	"sync"
+	"time"
+)
+
+type Queue struct {
+	mutex   sync.Mutex
+	getCond *sync.Cond
+	putCond *sync.Cond
+	list    *list.List
+}
+
+func NewQueue() *Queue {
+	queue := new(Queue)
+	queue.mutex = sync.Mutex{}
+	queue.getCond = sync.NewCond(&queue.mutex)
+	queue.putCond = sync.NewCond(&queue.mutex)
+	queue.list = list.New()
+	return queue
+}
+
+func (queue *Queue) Size() int {
+	queue.mutex.Lock()
+	size := queue.list.Len()
+	queue.mutex.Unlock()
+	return size
+}
+
+func (queue *Queue) IsEmpty() bool {
+	queue.mutex.Lock()
+	isEmpty := queue.isEmpty()
+	queue.mutex.Unlock()
+	return isEmpty
+}
+
+func (queue *Queue) Get(block bool, timeout float64) (*list.Element, error) {
+	queue.getCond.L.Lock()
+	defer queue.getCond.L.Unlock()
+	emptyQ := false
+	if !block {
+		if queue.isEmpty() {
+			emptyQ = true
+		}
+	} else if timeout == float64(0) {
+		for queue.isEmpty() {
+			queue.getCond.Wait()
+		}
+	} else if timeout < float64(0) {
+		return &list.Element{}, errors.New("'timeout' must be a non-negative number")
+	} else {
+		timer := time.After(time.Duration(timeout) * time.Second)
+		notEmpty := make(chan bool)
+		go queue.wait(notEmpty)
+		for queue.isEmpty() {
+			select {
+			case <-timer:
+				emptyQ = true
+				break
+			case <-notEmpty:
+				if queue.isEmpty() {
+					go queue.wait(notEmpty)
+				} else {
+					break
+				}
+			}
+		}
+	}
+	if emptyQ {
+		return &list.Element{}, errors.New("Empty Queue")
+	}
+	e := queue.list.Front()
+	queue.list.Remove(e)
+	return e, nil
+}
+
+func (queue *Queue) Put(element interface{}) *list.Element {
+	queue.putCond.L.Lock()
+	defer queue.putCond.L.Unlock()
+	e := queue.list.PushBack(element)
+	queue.getCond.Signal()
+	return e
+}
+
+func (queue *Queue) wait(c chan<- bool) {
+	queue.getCond.Wait()
+	c <- true
+}
+
+func (queue *Queue) isEmpty() bool {
+	return (queue.list.Len() == 0)
+}

queue_test.go

@@ -0,0 +1,66 @@
+package goqueue
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestUnBlockGet(t *testing.T) {
+	value := 8888
+
+	fmt.Println("Test unblock get, should no error...")
+	queue := NewQueue()
+	queue.Put(value)
+	e, err := queue.Get(false, 0)
+	if err != nil {
+		t.Fatalf("Unexpect error: %v\n", err)
+	} else if e.Value.(int) != value {
+		t.Fatalf("Expect %v, got %v\n", value, e.Value.(int))
+	}
+	fmt.Println("  ...PASSED")
+
+	fmt.Println("Test unblock get, error should show up...")
+	emptyQ := NewQueue()
+	_, err1 := emptyQ.Get(false, 0)
+	if err1 == nil {
+		t.Fatalf("No error show up\n")
+	}
+	fmt.Println("  ...PASSED")
+}
+
+func TestBlockGetWithTimeout(t *testing.T) {
+	w := make(chan bool)
+	queue := NewQueue()
+
+	get := func(timeout float64, value int, noErr bool) {
+		e, err := queue.Get(true, timeout)
+		if noErr {
+			if err == nil && e.Value.(int) != value {
+				t.Fatalf("Expect %v, got %v\n", value, e.Value.(int))
+			} else if err != nil {
+				t.Fatalf("Unexpect error: %v\n", err)
+			}
+		} else if err == nil {
+			t.Fatalf("Wanted an error, but got nil\n")
+		}
+		w <- true
+	}
+
+	put := func(d time.Duration, value int) {
+		time.Sleep(time.Second * d)
+		queue.Put(value)
+	}
+
+	fmt.Println("Test block get with timeout, should no error...")
+	go put(time.Duration(2), 8888)
+	go get(3, 8888, true)
+	<-w
+	fmt.Println("  ...PASSED")
+
+	fmt.Println("Test block get with timeout, error should show up...")
+	go put(time.Duration(4), 8888)
+	go get(3, 8888, false)
+	<-w
+	fmt.Println("  ...PASSED")
+}