Merge pull request #18 from wongoo/feature-demo

Add: getty demo

demo/hello/README.md

+# Run Hello Demo
+
+## 1. prepare 
+```bash
+
+git clone https://github.com/dubbogo/getty.git
+
+cd getty/demo/hello
+```
+
+## 2. run server
+
+run server: 
+`go run tcp/server/server.go`
+
+Or run server in task pool mode:
+```bash
+go run tcp/server/server.go -taskPool=true \
+    -task_queue_length=100 \
+    -task_queue_number=4 \
+    -task_pool_size=2000
+```
+
+## 3. run client
+
+```bash
+go run tcp/client/client.go
+```
\ No newline at end of file

demo/hello/consts.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package hello
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+const (
+	CronPeriod      = 20e9
+	WritePkgTimeout = 1e8
+)
+
+var (
+	log = getty.GetLogger()
+)

demo/hello/hello.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package hello
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+var (
+	Sessions []getty.Session
+)
+
+func ClientRequest() {
+	for _, session := range Sessions {
+		go func() {
+			echoTimes := 10
+			for i := 0; i < echoTimes; i++ {
+				err := session.WritePkg("hello", WritePkgTimeout)
+				if err != nil {
+					log.Infof("session.WritePkg(session{%s}, error{%v}", session.Stat(), err)
+					session.Close()
+				}
+			}
+			log.Infof("after loop %d times", echoTimes)
+		}()
+	}
+
+}

demo/hello/messagelistener.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package hello
+
+import (
+	"time"
+)
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+type MessageHandler struct {
+	SessionOnOpen func(session getty.Session)
+}
+
+func (h *MessageHandler) OnOpen(session getty.Session) error {
+	log.Infof("OnOpen session{%s} open", session.Stat())
+	if h.SessionOnOpen != nil {
+		h.SessionOnOpen(session)
+	}
+	return nil
+}
+
+func (h *MessageHandler) OnError(session getty.Session, err error) {
+	log.Infof("OnError session{%s} got error{%v}, will be closed.", session.Stat(), err)
+}
+
+func (h *MessageHandler) OnClose(session getty.Session) {
+	log.Infof("OnClose session{%s} is closing......", session.Stat())
+}
+
+func (h *MessageHandler) OnMessage(session getty.Session, pkg interface{}) {
+	s, ok := pkg.(string)
+	if !ok {
+		log.Infof("illegal package{%#v}", pkg)
+		return
+	}
+
+	log.Infof("OnMessage: %s", s)
+}
+
+func (h *MessageHandler) OnCron(session getty.Session) {
+	active := session.GetActive()
+	if CronPeriod < time.Since(active).Nanoseconds() {
+		log.Infof("OnCorn session{%s} timeout{%s}", session.Stat(), time.Since(active).String())
+		session.Close()
+	}
+}

demo/hello/pkghandler.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package hello
+
+import (
+	"errors"
+)
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+type PackageHandler struct{}
+
+func (h *PackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
+	s := string(data)
+	return s, len(s), nil
+}
+
+func (h *PackageHandler) Write(ss getty.Session, pkg interface{}) error {
+	s, ok := pkg.(string)
+	if !ok {
+		log.Infof("illegal pkg:%+v", pkg)
+		return errors.New("invalid package")
+	}
+	return ss.WriteBytes([]byte(s))
+}

demo/hello/tcp/client/client.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package main
+
+import (
+	"flag"
+)
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+import (
+	"github.com/dubbogo/getty/demo/hello"
+	"github.com/dubbogo/getty/demo/hello/tcp"
+	"github.com/dubbogo/getty/demo/util"
+)
+
+var (
+	ip          = flag.String("ip", "127.0.0.1", "server IP")
+	connections = flag.Int("conn", 1, "number of tcp connections")
+)
+
+func main() {
+	flag.Parse()
+	util.SetLimit()
+
+	client := getty.NewTCPClient(
+		getty.WithServerAddress(*ip+":8090"),
+		getty.WithConnectionNumber(*connections),
+	)
+
+	client.RunEventLoop(tcp.NewHelloClientSession)
+
+	go hello.ClientRequest()
+
+	util.WaitCloseSignals(client)
+}

demo/hello/tcp/config.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package tcp
+
+import (
+	"fmt"
+	"net"
+	"time"
+)
+
+import (
+	"github.com/dubbogo/getty"
+)
+
+import (
+	"github.com/dubbogo/getty/demo/hello"
+)
+
+var (
+	pkgHandler    = &hello.PackageHandler{}
+	eventListener = &hello.MessageHandler{}
+)
+
+func NewHelloClientSession(session getty.Session) (err error) {
+	eventListener.SessionOnOpen = func(session getty.Session) {
+		hello.Sessions = append(hello.Sessions, session)
+	}
+	return InitialSession(session)
+}
+
+func InitialSession(session getty.Session) (err error) {
+	session.SetCompressType(getty.CompressZip)
+
+	tcpConn, ok := session.Conn().(*net.TCPConn)
+	if !ok {
+		panic(fmt.Sprintf("newSession: %s, session.conn{%#v} is not tcp connection", session.Stat(), session.Conn()))
+	}
+
+	if err = tcpConn.SetNoDelay(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlive(true); err != nil {
+		return err
+	}
+	if err = tcpConn.SetKeepAlivePeriod(10 * time.Second); err != nil {
+		return err
+	}
+	if err = tcpConn.SetReadBuffer(262144); err != nil {
+		return err
+	}
+	if err = tcpConn.SetWriteBuffer(524288); err != nil {
+		return err
+	}
+
+	session.SetName("hello")
+	session.SetMaxMsgLen(128)
+	session.SetRQLen(1024)
+	session.SetWQLen(512)
+	session.SetReadTimeout(time.Second)
+	session.SetWriteTimeout(5 * time.Second)
+	session.SetCronPeriod(int(hello.CronPeriod / 1e6))
+	session.SetWaitTime(time.Second)
+
+	session.SetPkgHandler(pkgHandler)
+	session.SetEventListener(eventListener)
+	return nil
+}

demo/hello/tcp/server/server.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package main
+
+import (
+	"flag"
+)
+
+import (
+	"github.com/dubbogo/getty"
+	gxsync "github.com/dubbogo/gost/sync"
+)
+
+import (
+	"github.com/dubbogo/getty/demo/hello/tcp"
+	"github.com/dubbogo/getty/demo/util"
+)
+
+var (
+	taskPollMode        = flag.Bool("taskPool", false, "task pool mode")
+	taskPollQueueLength = flag.Int("task_queue_length", 100, "task queue length")
+	taskPollQueueNumber = flag.Int("task_queue_number", 4, "task queue number")
+	taskPollSize        = flag.Int("task_pool_size", 2000, "task poll size")
+)
+
+var (
+	taskPoll *gxsync.TaskPool
+)
+
+func main() {
+	flag.Parse()
+
+	util.SetLimit()
+
+	options := []getty.ServerOption{getty.WithLocalAddress(":8090")}
+
+	if *taskPollMode {
+		taskPoll = gxsync.NewTaskPool(
+			gxsync.WithTaskPoolTaskQueueLength(*taskPollQueueLength),
+			gxsync.WithTaskPoolTaskQueueNumber(*taskPollQueueNumber),
+			gxsync.WithTaskPoolTaskPoolSize(*taskPollSize),
+		)
+	}
+
+	server := getty.NewTCPServer(options...)
+
+	go server.RunEventLoop(NewHelloServerSession)
+
+	util.WaitCloseSignals(server)
+}
+
+func NewHelloServerSession(session getty.Session) (err error) {
+	err = tcp.InitialSession(session)
+	if err != nil {
+		return
+	}
+	session.SetTaskPool(taskPoll)
+	return
+}

demo/util/limit_linux.go

+// +build linux
+
+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package util
+
+import (
+	"syscall"
+)
+
+func SetLimit() {
+	var rLimit syscall.Rlimit
+	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil {
+		panic(err)
+	}
+	rLimit.Cur = rLimit.Max
+	if err := syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil {
+		panic(err)
+	}
+}

demo/util/limit_other.go

+// +build !linux
+
+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package util
+
+func SetLimit() {
+}

demo/util/signals.go

+/******************************************************
+# MAINTAINER : wongoo
+# LICENCE    : Apache License 2.0
+# EMAIL      : gelnyang@163.com
+# MOD        : 2019-06-11
+******************************************************/
+
+package util
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+)
+
+type Closer interface {
+	Close()
+}
+
+func WaitCloseSignals(closer Closer) {
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, os.Interrupt, os.Kill, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT)
+	<-signals
+	closer.Close()
+}

logger.go

@@ -85,6 +85,11 @@ func SetLogger(logger Logger) {
 	log = logger
 }
 
+// GetLogger get getty logger
+func GetLogger() Logger {
+	return log
+}
+
 // SetLoggerLevel
 func SetLoggerLevel(level LoggerLevel) error {
 	var err error