Merge branch 'master' of https://github.com/duanhf2012/origin

cluster/cluster.go

@@ -116,6 +116,9 @@ func (slf *CCluster) ConnService() error {
 	//连接集群服务器
 	for _, nodeList := range slf.cfg.mapClusterNodeService {
 		for _, node := range nodeList {
+			if node.NodeID == slf.cfg.currentNode.NodeID {
+				continue
+			}
 			slf.nodeclient[node.NodeID] = &RpcClient{node.NodeID, nil, node.ServerAddr, false}
 		}
 	}

util/aes_encrypt.go

@@ -0,0 +1,89 @@
+// Copyright 2014 mqantserver Author. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package util
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"fmt"
+)
+
+func NewAesEncrypt(key string) (aes *AesEncrypt, err error) {
+	keyLen := len(key)
+	if keyLen < 16 {
+		err = fmt.Errorf("The length of res key shall not be less than 16")
+		return
+	}
+	aes = &AesEncrypt{
+		StrKey: key,
+	}
+	return aes, nil
+}
+
+type AesEncrypt struct {
+	StrKey string
+}
+
+func (this *AesEncrypt) getKey() []byte {
+	keyLen := len(this.StrKey)
+	if keyLen < 16 {
+		panic("The length of res key shall not be less than 16")
+	}
+	arrKey := []byte(this.StrKey)
+	if keyLen >= 32 {
+		//取前32个字节
+		return arrKey[:32]
+	}
+	if keyLen >= 24 {
+		//取前24个字节
+		return arrKey[:24]
+	}
+	//取前16个字节
+	return arrKey[:16]
+}
+
+//加密字符串
+func (this *AesEncrypt) Encrypt(strMesg string) ([]byte, error) {
+	key := this.getKey()
+	var iv = []byte(key)[:aes.BlockSize]
+	encrypted := make([]byte, len(strMesg))
+	aesBlockEncrypter, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	aesEncrypter := cipher.NewCFBEncrypter(aesBlockEncrypter, iv)
+	aesEncrypter.XORKeyStream(encrypted, []byte(strMesg))
+	return encrypted, nil
+}
+
+//解密字符串
+func (this *AesEncrypt) Decrypt(src []byte) (strDesc string, err error) {
+	defer func() {
+		//错误处理
+		if e := recover(); e != nil {
+			err = e.(error)
+		}
+	}()
+	key := this.getKey()
+	var iv = []byte(key)[:aes.BlockSize]
+	decrypted := make([]byte, len(src))
+	var aesBlockDecrypter cipher.Block
+	aesBlockDecrypter, err = aes.NewCipher([]byte(key))
+	if err != nil {
+		return "", err
+	}
+	aesDecrypter := cipher.NewCFBDecrypter(aesBlockDecrypter, iv)
+	aesDecrypter.XORKeyStream(decrypted, src)
+	return string(decrypted), nil
+}

util/queue.go

@@ -0,0 +1,101 @@
+/*
+Package queue provides a fast, ring-buffer queue based on the version suggested by Dariusz Górecki.
+Using this instead of other, simpler, queue implementations (slice+append or linked list) provides
+substantial memory and time benefits, and fewer GC pauses.
+The queue implemented here is as fast as it is for an additional reason: it is *not* thread-safe.
+*/
+package util
+
+// minQueueLen is smallest capacity that queue may have.
+// Must be power of 2 for bitwise modulus: x % n == x & (n - 1).
+const minQueueLen = 16
+
+// Queue represents a single instance of the queue data structure.
+type Queue struct {
+	buf               []interface{}
+	head, tail, count int
+}
+
+// New constructs and returns a new Queue.
+func NewQueue() *Queue {
+	return &Queue{
+		buf: make([]interface{}, minQueueLen),
+	}
+}
+
+// Length returns the number of elements currently stored in the queue.
+func (q *Queue) Length() int {
+	return q.count
+}
+
+// resizes the queue to fit exactly twice its current contents
+// this can result in shrinking if the queue is less than half-full
+func (q *Queue) resize() {
+	newBuf := make([]interface{}, q.count<<1)
+
+	if q.tail > q.head {
+		copy(newBuf, q.buf[q.head:q.tail])
+	} else {
+		n := copy(newBuf, q.buf[q.head:])
+		copy(newBuf[n:], q.buf[:q.tail])
+	}
+
+	q.head = 0
+	q.tail = q.count
+	q.buf = newBuf
+}
+
+// Add puts an element on the end of the queue.
+func (q *Queue) Add(elem interface{}) {
+	if q.count == len(q.buf) {
+		q.resize()
+	}
+
+	q.buf[q.tail] = elem
+	// bitwise modulus
+	q.tail = (q.tail + 1) & (len(q.buf) - 1)
+	q.count++
+}
+
+// Peek returns the element at the head of the queue. This call panics
+// if the queue is empty.
+func (q *Queue) Peek() interface{} {
+	if q.count <= 0 {
+		panic("queue: Peek() called on empty queue")
+	}
+	return q.buf[q.head]
+}
+
+// Get returns the element at index i in the queue. If the index is
+// invalid, the call will panic. This method accepts both positive and
+// negative index values. Index 0 refers to the first element, and
+// index -1 refers to the last.
+func (q *Queue) Get(i int) interface{} {
+	// If indexing backwards, convert to positive index.
+	if i < 0 {
+		i += q.count
+	}
+	if i < 0 || i >= q.count {
+		panic("queue: Get() called with index out of range")
+	}
+	// bitwise modulus
+	return q.buf[(q.head+i)&(len(q.buf)-1)]
+}
+
+// Remove removes and returns the element from the front of the queue. If the
+// queue is empty, the call will panic.
+func (q *Queue) Remove() interface{} {
+	if q.count <= 0 {
+		panic("queue: Remove() called on empty queue")
+	}
+	ret := q.buf[q.head]
+	q.buf[q.head] = nil
+	// bitwise modulus
+	q.head = (q.head + 1) & (len(q.buf) - 1)
+	q.count--
+	// Resize down if buffer 1/4 full.
+	if len(q.buf) > minQueueLen && (q.count<<2) == len(q.buf) {
+		q.resize()
+	}
+	return ret
+}

util/sync_queue.go

@@ -0,0 +1,96 @@
+package util
+
+import (
+	"sync"
+)
+
+// Synchronous FIFO queue
+type SyncQueue struct {
+	lock    sync.Mutex
+	popable *sync.Cond
+	buffer  *Queue
+	closed  bool
+}
+
+// Create a new SyncQueue
+func NewSyncQueue() *SyncQueue {
+	ch := &SyncQueue{
+		buffer: NewQueue(),
+	}
+	ch.popable = sync.NewCond(&ch.lock)
+	return ch
+}
+
+// Pop an item from SyncQueue, will block if SyncQueue is empty
+func (q *SyncQueue) Pop() (v interface{}) {
+	c := q.popable
+	buffer := q.buffer
+
+	q.lock.Lock()
+	for buffer.Length() == 0 && !q.closed {
+		c.Wait()
+	}
+
+	if buffer.Length() > 0 {
+		v = buffer.Peek()
+		buffer.Remove()
+	}
+
+	q.lock.Unlock()
+	return
+}
+
+// Try to pop an item from SyncQueue, will return immediately with bool=false if SyncQueue is empty
+func (q *SyncQueue) TryPop() (v interface{}, ok bool) {
+	buffer := q.buffer
+
+	q.lock.Lock()
+
+	if buffer.Length() > 0 {
+		v = buffer.Peek()
+		buffer.Remove()
+		ok = true
+	} else if q.closed {
+		ok = true
+	}
+
+	q.lock.Unlock()
+	return
+}
+
+// Push an item to SyncQueue. Always returns immediately without blocking
+func (q *SyncQueue) Push(v interface{}) {
+	q.lock.Lock()
+	if !q.closed {
+		q.buffer.Add(v)
+		q.popable.Signal()
+	}
+	q.lock.Unlock()
+}
+
+// Get the length of SyncQueue
+func (q *SyncQueue) Len() (l int) {
+	q.lock.Lock()
+	l = q.buffer.Length()
+	q.lock.Unlock()
+	return
+}
+
+// Close SyncQueue
+//
+// After close, Pop will return nil without block, and TryPop will return v=nil, ok=True
+func (q *SyncQueue) Close() {
+	q.lock.Lock()
+	if !q.closed {
+		q.closed = true
+		q.popable.Signal()
+	}
+	q.lock.Unlock()
+}
+
+func (q *SyncQueue) IsClose() (v bool) {
+	q.lock.Lock()
+	v = q.closed
+	q.lock.Unlock()
+	return
+}

util/uuid.go

@@ -0,0 +1,111 @@
+// Copyright 2014 mqant Author. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package util
+
+import (
+	crand "crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	mrand "math/rand"
+	"regexp"
+	"strings"
+	"time"
+)
+
+// seeded indicates if math/rand has been seeded
+var seeded bool = false
+
+// uuidRegex matches the UUID string
+var uuidRegex *regexp.Regexp = regexp.MustCompile(`^\{?([a-fA-F0-9]{8})-?([a-fA-F0-9]{4})-?([a-fA-F0-9]{4})-?([a-fA-F0-9]{4})-?([a-fA-F0-9]{12})\}?$`)
+
+// UUID type.
+type UUID [16]byte
+
+// Hex returns a hex string representation of the UUID in xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx format.
+func (this UUID) Hex() string {
+	x := [16]byte(this)
+	return fmt.Sprintf("%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x",
+		x[0], x[1], x[2], x[3], x[4],
+		x[5], x[6],
+		x[7], x[8],
+		x[9], x[10], x[11], x[12], x[13], x[14], x[15])
+
+}
+
+// Rand generates a new version 4 UUID.
+func Rand() UUID {
+	var x [16]byte
+	randBytes(x[:])
+	x[6] = (x[6] & 0x0F) | 0x40
+	x[8] = (x[8] & 0x3F) | 0x80
+	return x
+}
+
+// FromStr returns a UUID based on a string.
+// The string could be in the following format:
+//
+// xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+//
+// xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+//
+// {xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx}
+//
+// If the string is not in one of these formats, it'll return an error.
+func FromStr(s string) (id UUID, err error) {
+	if s == "" {
+		err = errors.New("Empty string")
+		return
+	}
+
+	parts := uuidRegex.FindStringSubmatch(s)
+	if parts == nil {
+		err = errors.New("Invalid string format")
+		return
+	}
+
+	var array [16]byte
+	slice, _ := hex.DecodeString(strings.Join(parts[1:], ""))
+	copy(array[:], slice)
+	id = array
+	return
+}
+
+// MustFromStr behaves similarly to FromStr except that it'll panic instead of
+// returning an error.
+func MustFromStr(s string) UUID {
+	id, err := FromStr(s)
+	if err != nil {
+		panic(err)
+	}
+	return id
+}
+
+// randBytes uses crypto random to get random numbers. If fails then it uses math random.
+func randBytes(x []byte) {
+
+	length := len(x)
+	n, err := crand.Read(x)
+
+	if n != length || err != nil {
+		if !seeded {
+			mrand.Seed(time.Now().UnixNano())
+		}
+
+		for length > 0 {
+			length--
+			x[length] = byte(mrand.Int31n(256))
+		}
+	}
+}