first commit

LICENSE

@@ -0,0 +1,15 @@
+
+            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+                    Version 2, December 2004
+
+ Copyright (C) 2004 Sam Hocevar <[email protected]>
+
+ Everyone is permitted to copy and distribute verbatim or modified
+ copies of this license document, and changing it is allowed as long
+ as the name is changed.
+
+            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+   TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
+
+  0. You just DO WHAT THE FUCK YOU WANT TO.
+

README.md

@@ -0,0 +1,21 @@
+### Introduction
+
+Simple Golang SECS/GEM implementation.
+
+* [x] SECS-II
+* [x] HSMS
+* [x] SML
+* [ ] GEM
+* [ ] HSMS-SS, HSMS-GS
+
+### Acknowledgements
+
+* [funny/link](https://github.com/funny/link)：Go Networking Scaffold
+* [wolimst/lib-secs2-hsms-go](https://github.com/wolimst/lib-secs2-hsms-go)：SECS-II/HSMS/SML Data Parser
+
+### Other
+
+The third-party library has been heavily modified, so it is directly imported and used in the project.
+
+This library is only a portion, completing the most fundamental part. Use it with caution.
+

api.go

@@ -0,0 +1,86 @@
+package link
+
+import (
+	"io"
+	"net"
+	"strings"
+	"time"
+)
+
+type Protocol interface {
+	NewCodec(rw io.ReadWriter) (Codec, error)
+}
+
+type ProtocolFunc func(rw io.ReadWriter) (Codec, error)
+
+func (pf ProtocolFunc) NewCodec(rw io.ReadWriter) (Codec, error) {
+	return pf(rw)
+}
+
+type Codec interface {
+	Receive() (interface{}, error)
+	Send(interface{}) error
+	Close() error
+}
+
+type ClearSendChan interface {
+	ClearSendChan(<-chan interface{})
+}
+
+func Listen(network, address string, protocol Protocol, sendChanSize int, handler Handler) (*Server, error) {
+	listener, err := net.Listen(network, address)
+	if err != nil {
+		return nil, err
+	}
+	return NewServer(listener, protocol, sendChanSize, handler), nil
+}
+
+func Dial(network, address string, protocol Protocol, sendChanSize int) (*Session, error) {
+	conn, err := net.Dial(network, address)
+	if err != nil {
+		return nil, err
+	}
+	codec, err := protocol.NewCodec(conn)
+	if err != nil {
+		return nil, err
+	}
+	return NewSession(codec, sendChanSize), nil
+}
+
+func DialTimeout(network, address string, timeout time.Duration, protocol Protocol, sendChanSize int) (*Session, error) {
+	conn, err := net.DialTimeout(network, address, timeout)
+	if err != nil {
+		return nil, err
+	}
+	codec, err := protocol.NewCodec(conn)
+	if err != nil {
+		return nil, err
+	}
+	return NewSession(codec, sendChanSize), nil
+}
+
+func Accept(listener net.Listener) (net.Conn, error) {
+	var tempDelay time.Duration
+	for {
+		conn, err := listener.Accept()
+		if err != nil {
+			if ne, ok := err.(net.Error); ok && ne.Temporary() {
+				if tempDelay == 0 {
+					tempDelay = 5 * time.Millisecond
+				} else {
+					tempDelay *= 2
+				}
+				if max := 1 * time.Second; tempDelay > max {
+					tempDelay = max
+				}
+				time.Sleep(tempDelay)
+				continue
+			}
+			if strings.Contains(err.Error(), "use of closed network connection") {
+				return nil, io.EOF
+			}
+			return nil, err
+		}
+		return conn, nil
+	}
+}

channel.go

@@ -0,0 +1,87 @@
+package link
+
+import (
+	"sync"
+)
+
+type KEY interface{}
+
+type Channel struct {
+	mutex    sync.RWMutex
+	sessions map[KEY]*Session
+
+	// channel state
+	State interface{}
+}
+
+func NewChannel() *Channel {
+	return &Channel{
+		sessions: make(map[KEY]*Session),
+	}
+}
+
+func (channel *Channel) Len() int {
+	channel.mutex.RLock()
+	defer channel.mutex.RUnlock()
+	return len(channel.sessions)
+}
+
+func (channel *Channel) Fetch(callback func(*Session)) {
+	channel.mutex.RLock()
+	defer channel.mutex.RUnlock()
+	for _, session := range channel.sessions {
+		callback(session)
+	}
+}
+
+func (channel *Channel) Get(key KEY) *Session {
+	channel.mutex.RLock()
+	defer channel.mutex.RUnlock()
+	session, _ := channel.sessions[key]
+	return session
+}
+
+func (channel *Channel) Put(key KEY, session *Session) {
+	channel.mutex.Lock()
+	defer channel.mutex.Unlock()
+	if session, exists := channel.sessions[key]; exists {
+		channel.remove(key, session)
+	}
+	session.AddCloseCallback(channel, key, func() {
+		channel.Remove(key)
+	})
+	channel.sessions[key] = session
+}
+
+func (channel *Channel) remove(key KEY, session *Session) {
+	session.RemoveCloseCallback(channel, key)
+	delete(channel.sessions, key)
+}
+
+func (channel *Channel) Remove(key KEY) bool {
+	channel.mutex.Lock()
+	defer channel.mutex.Unlock()
+	session, exists := channel.sessions[key]
+	if exists {
+		channel.remove(key, session)
+	}
+	return exists
+}
+
+func (channel *Channel) FetchAndRemove(callback func(*Session)) {
+	channel.mutex.Lock()
+	defer channel.mutex.Unlock()
+	for key, session := range channel.sessions {
+		session.RemoveCloseCallback(channel, key)
+		delete(channel.sessions, key)
+		callback(session)
+	}
+}
+
+func (channel *Channel) Close() {
+	channel.mutex.Lock()
+	defer channel.mutex.Unlock()
+	for key, session := range channel.sessions {
+		channel.remove(key, session)
+	}
+}

codec/bufio.go

@@ -0,0 +1,93 @@
+package codec
+
+import (
+	"bufio"
+	link "github.com/younglifestyle/secs4go"
+	"io"
+)
+
+func Bufio(base link.Protocol, readBuf, writeBuf int) link.Protocol {
+	return &bufioProtocol{
+		base:     base,
+		readBuf:  readBuf,
+		writeBuf: writeBuf,
+	}
+}
+
+type bufioProtocol struct {
+	base     link.Protocol
+	readBuf  int
+	writeBuf int
+}
+
+func (b *bufioProtocol) NewCodec(rw io.ReadWriter) (cc link.Codec, err error) {
+	codec := new(bufioCodec)
+
+	if b.writeBuf > 0 {
+		codec.stream.w = bufio.NewWriterSize(rw, b.writeBuf)
+		codec.stream.Writer = codec.stream.w
+	} else {
+		codec.stream.Writer = rw
+	}
+
+	if b.readBuf > 0 {
+		codec.stream.Reader = bufio.NewReaderSize(rw, b.readBuf)
+	} else {
+		codec.stream.Reader = rw
+	}
+
+	codec.stream.c, _ = rw.(io.Closer)
+
+	codec.base, err = b.base.NewCodec(&codec.stream)
+	if err != nil {
+		return
+	}
+	cc = codec
+	return
+}
+
+type bufioStream struct {
+	io.Reader
+	io.Writer
+	c io.Closer
+	w *bufio.Writer
+}
+
+func (s *bufioStream) Flush() error {
+	if s.w != nil {
+		return s.w.Flush()
+	}
+	return nil
+}
+
+func (s *bufioStream) close() error {
+	if s.c != nil {
+		return s.c.Close()
+	}
+	return nil
+}
+
+type bufioCodec struct {
+	base   link.Codec
+	stream bufioStream
+}
+
+func (c *bufioCodec) Send(msg interface{}) error {
+	if err := c.base.Send(msg); err != nil {
+		return err
+	}
+	return c.stream.Flush()
+}
+
+func (c *bufioCodec) Receive() (interface{}, error) {
+	return c.base.Receive()
+}
+
+func (c *bufioCodec) Close() error {
+	err1 := c.base.Close()
+	err2 := c.stream.close()
+	if err1 != nil {
+		return err1
+	}
+	return err2
+}

codec/bufio_test.go

@@ -0,0 +1,10 @@
+package codec
+
+import (
+	"encoding/binary"
+	"testing"
+)
+
+func Test_Bufio(t *testing.T) {
+	JsonTest(t, Bufio(FixLen(JsonTestProtocol(), 2, binary.LittleEndian, 64*1024, 64*1024), 1024, 1024))
+}

codec/byte.go

@@ -0,0 +1,63 @@
+package codec
+
+import (
+	"errors"
+	link "github.com/younglifestyle/secs4go"
+	"io"
+)
+
+type ByteProtocol struct {
+	//data []byte
+}
+
+func (b *ByteProtocol) NewCodec(rw io.ReadWriter) (link.Codec, error) {
+	codec := &byteCodec{
+		p: b,
+		r: rw,
+		w: rw,
+	}
+
+	codec.closer, _ = rw.(io.Closer)
+	return codec, nil
+}
+
+func Byte() *ByteProtocol {
+	return &ByteProtocol{
+		//data: make([]byte, 0),
+	}
+}
+
+type byteCodec struct {
+	r      io.Reader
+	w      io.Writer
+	p      *ByteProtocol
+	closer io.Closer
+}
+
+func (c *byteCodec) Receive() (interface{}, error) {
+
+	recvData := make([]byte, 4092)
+
+	cnt, err := c.r.Read(recvData)
+
+	return recvData[:cnt], err
+}
+
+func (c *byteCodec) Send(msg interface{}) error {
+
+	b, ok := msg.([]byte)
+	if !ok {
+		return errors.New("Send Byte Format Error")
+	}
+
+	_, err := c.w.Write(b)
+
+	return err
+}
+
+func (c *byteCodec) Close() error {
+	if c.closer != nil {
+		return c.closer.Close()
+	}
+	return nil
+}