dialog expires should be from Expires header and default expires sho…

…uld be configurable #13

backend.go

@@ -372,6 +372,8 @@ type DialogBasedBackend struct {
 }
 
 func NewDialogBasedBackend(timeoutSeconds int64) *DialogBasedBackend {
+	zap.L().Info("set the dialog timeout ", zap.Int64("timeout", timeoutSeconds))
+
 	return &DialogBasedBackend{timeout: time.Duration(timeoutSeconds) * time.Second,
 		backends:      make(map[string]*ExpireBackend),
 		nextCleanTime: time.Now().Add(time.Duration(timeoutSeconds) * time.Second)}
@@ -388,8 +390,12 @@ func (dbb *DialogBasedBackend) GetBackend(dialog string) (Backend, error) {
 
 }
 
-func (dbb *DialogBasedBackend) AddBackend(dialog string, backend Backend) {
-	expire := time.Now().Add(dbb.timeout)
+func (dbb *DialogBasedBackend) AddBackend(dialog string, backend Backend, expireSeconds int) {
+	timeout := dbb.timeout
+	if float64(expireSeconds) > timeout.Seconds() {
+		timeout = time.Duration(expireSeconds) * time.Second
+	}
+	expire := time.Now().Add(timeout)
 	dbb.backends[dialog] = &ExpireBackend{backend: backend, expire: expire}
 	if dbb.nextCleanTime.Before(time.Now()) {
 		dbb.nextCleanTime = expire

main.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -21,8 +22,9 @@ type HostIp struct {
 }
 
 type ProxyConfig struct {
-	Name    string
-	Listens []struct {
+	Name          string
+	DialogTimeout int `yaml:"dialogTimeout,omitempty"`
+	Listens       []struct {
 		Address    string
 		UDPPort    int      `yaml:"udp-port,omitempty"`
 		TCPPort    int      `yaml:"tcp-port,omitempty"`
@@ -139,9 +141,24 @@ func startProxies(c *cli.Context) error {
 	}
 }
 
+func getDefaultDialogTimeout() int {
+	expire, ok := os.LookupEnv("DEFAULT_DIALOG_TIMEOUT")
+	if !ok {
+		return 1200
+	}
+	if val, err := strconv.Atoi(expire); err == nil {
+		return val
+	}
+	return 1200
+}
+
 func startProxy(config ProxyConfig, preConfigRoute *PreConfigRoute, resolver *PreConfigHostResolver) error {
 	selfLearnRoute := NewSelfLearnRoute()
-	proxy := NewProxy(config.Name, preConfigRoute, resolver, selfLearnRoute)
+	dialogTimeout := config.DialogTimeout
+	if dialogTimeout <= 0 {
+		dialogTimeout = getDefaultDialogTimeout()
+	}
+	proxy := NewProxy(config.Name, int64(dialogTimeout), preConfigRoute, resolver, selfLearnRoute)
 	for _, listen := range config.Listens {
 		item, err := NewProxyItem(listen.Address,
 			listen.UDPPort,

message.go

@@ -588,7 +588,6 @@ func (m *Message) IsResponse() bool {
 	return m.response != nil
 }
 
-//
 // All 2xx, 3xx, 4xx, 5xx and 6xx responses are final.
 func (m *Message) IsFinalResponse() bool {
 	if m.response == nil {
@@ -794,3 +793,12 @@ func (m *Message) GetServerTransaction() (string, error) {
 
 	return fmt.Sprintf("%s-%s-%s", m.request.method, sentBy, branch), nil
 }
+
+// Get the value of Expires
+func (m *Message) GetExpires(defValue int) int {
+	expires, err := m.GetHeaderInt("Expires")
+	if err == nil {
+		return expires
+	}
+	return defValue
+}

proxy.go

@@ -129,6 +129,7 @@ type Proxy struct {
 }
 
 func NewProxy(name string,
+	dialogExpire int64,
 	preConfigRoute *PreConfigRoute,
 	resolver *PreConfigHostResolver,
 	selfLearnRoute *SelfLearnRoute) *Proxy {
@@ -142,7 +143,7 @@ func NewProxy(name string,
 		backendChangeChannel: make(chan *BackendChangeEvent, 1000),
 		connAcceptedChannel:  make(chan net.Conn),
 		backends:             make(map[string]*BackendWithParent),
-		dialogBasedBackends:  NewDialogBasedBackend(600)}
+		dialogBasedBackends:  NewDialogBasedBackend(dialogExpire)}
 	go proxy.receiveAndProcessMessage()
 	return proxy
 }
@@ -275,7 +276,7 @@ func (p *Proxy) handleDialog(peerAddr string, peerPort int, msg *Message) {
 			dialog, _ := msg.GetDialog()
 			if dialog != "" {
 				zap.L().Info("dialog is bind to backend", zap.String("backendAddr", addr), zap.String("dialog", dialog))
-				p.dialogBasedBackends.AddBackend(dialog, backend)
+				p.dialogBasedBackends.AddBackend(dialog, backend, msg.GetExpires(0))
 			}
 		case "BYE":
 			dialog, _ := msg.GetDialog()
@@ -319,7 +320,7 @@ func (p *Proxy) HandleMessage(msg *Message) {
 				if dialog, err := msg.GetDialog(); err == nil {
 					zap.L().Info("bind the dialog to the response", zap.String("dialog", dialog), zap.String("backend", backendWithParent.backend.GetAddress()))
 
-					p.dialogBasedBackends.AddBackend(dialog, backendWithParent.backend)
+					p.dialogBasedBackends.AddBackend(dialog, backendWithParent.backend, msg.GetExpires(0))
 				}
 			}
 		}
@@ -387,7 +388,7 @@ func (p *Proxy) sendToBackend(msg *Message) {
 			transId, err := msg.GetClientTransaction()
 			if err == nil {
 				zap.L().Debug("bind client transaction with backend", zap.String("trandId", transId), zap.String("backend", backend.GetAddress()))
-				p.dialogBasedBackends.AddBackend(transId, backend)
+				p.dialogBasedBackends.AddBackend(transId, backend, msg.GetExpires(0))
 			}
 		}
 	}