mirror of
https://github.com/openziti/ziti.git
synced 2026-10-15 00:22:01 +00:00
@@ -57,6 +57,7 @@ func (self *baseRequestHandler) getChannel() channel2.Channel {
|
||||
}
|
||||
|
||||
func (self *baseRequestHandler) returnError(ctx requestContext, err controllerError) {
|
||||
ctx.CleanupOnError()
|
||||
responseMsg := channel2.NewMessage(int32(edge_ctrl_pb.ContentType_ErrorType), []byte(err.Error()))
|
||||
responseMsg.PutUint32Header(edge.ErrorCodeHeader, err.ErrorCode())
|
||||
responseMsg.ReplyTo(ctx.GetMessage())
|
||||
@@ -96,6 +97,7 @@ func (self *baseRequestHandler) logResult(ctx requestContext, err error) {
|
||||
type requestContext interface {
|
||||
GetHandler() requestHandler
|
||||
GetMessage() *channel2.Message
|
||||
CleanupOnError()
|
||||
}
|
||||
|
||||
type sessionRequestContext interface {
|
||||
@@ -110,6 +112,19 @@ type baseSessionRequestContext struct {
|
||||
sourceRouter *network.Router
|
||||
session *model.Session
|
||||
service *model.Service
|
||||
newSession bool
|
||||
}
|
||||
|
||||
func (self *baseSessionRequestContext) CleanupOnError() {
|
||||
if self.newSession && self.session != nil {
|
||||
logger := logrus.
|
||||
WithField("operation", self.handler.Label()).
|
||||
WithField("router", self.sourceRouter.Name)
|
||||
|
||||
if err := self.handler.getAppEnv().Handlers.Session.Delete(self.session.Id); err != nil {
|
||||
logger.WithError(err).Error("unable to delete session created before error encountered")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *baseSessionRequestContext) GetMessage() *channel2.Message {
|
||||
|
||||
@@ -85,7 +85,7 @@ func (self *baseTunnelRequestContext) ensureApiSession(configTypes []string) boo
|
||||
self.apiSession = apiSession
|
||||
|
||||
if _, err := self.handler.getAppEnv().GetHandlers().ApiSession.MarkActivityByTokens(self.apiSession.Token); err != nil {
|
||||
logger.WithError(err).Error("unexepcted error while marking api session activity")
|
||||
logger.WithError(err).Error("unexpected error while marking api session activity")
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -180,7 +180,9 @@ func (self *baseTunnelRequestContext) ensureSessionForService(sessionId, session
|
||||
self.session, err = self.handler.getAppEnv().Handlers.Session.Read(id)
|
||||
if err != nil {
|
||||
err = internalError(err)
|
||||
return
|
||||
}
|
||||
self.newSession = true
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -79,11 +79,11 @@ func (self *listTunnelServicesHandler) listServices(ctx *listTunnelServicesReque
|
||||
var lastUpdate time.Time
|
||||
if val, found := self.appEnv.IdentityRefreshMap.Get(ctx.identity.Id); found {
|
||||
lastUpdate = val.(time.Time)
|
||||
} else if lastUpdate.Before(self.appEnv.StartupTime) {
|
||||
} else {
|
||||
lastUpdate = self.appEnv.StartupTime
|
||||
}
|
||||
|
||||
if !ctx.lastUpdate.IsZero() && !ctx.lastUpdate.Before(lastUpdate) {
|
||||
if lastUpdate.Before(ctx.lastUpdate) {
|
||||
logger.Debug("service list requested, but no update available")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -53,9 +53,7 @@ type fabricProvider struct {
|
||||
tunneler *tunneler
|
||||
|
||||
apiSessionLock sync.Mutex
|
||||
apiSessionId string
|
||||
apiSessionToken string
|
||||
refreshInterval time.Duration
|
||||
currentIdentity *edge.CurrentIdentity
|
||||
|
||||
dialSessions cmap.ConcurrentMap
|
||||
@@ -84,9 +82,7 @@ func (self *fabricProvider) updateApiSession(resp *edge_ctrl_pb.CreateApiSession
|
||||
|
||||
self.tunneler.stateManager.RemoveConnectedApiSession(self.apiSessionToken)
|
||||
|
||||
self.apiSessionId = resp.SessionId
|
||||
self.apiSessionToken = resp.Token
|
||||
self.refreshInterval = time.Duration(resp.RefreshIntervalSeconds) * time.Second
|
||||
self.currentIdentity = &edge.CurrentIdentity{
|
||||
Id: resp.IdentityId,
|
||||
Name: resp.IdentityName,
|
||||
|
||||
Reference in New Issue
Block a user