mirror of
https://github.com/openziti/ziti.git
synced 2026-10-13 07:41:17 +00:00
Merge branch 'main' into transwarp_2
This commit is contained in:
@@ -133,13 +133,15 @@ func (entity *Terminator) SetValues(ctx *boltz.PersistContext) {
|
||||
ctx.SetInt32(FieldTerminatorCost, int32(entity.Cost))
|
||||
ctx.SetRequiredString(FieldTerminatorPrecedence, entity.Precedence)
|
||||
|
||||
_ = ctx.Bucket.DeleteBucket([]byte(FieldServerPeerData))
|
||||
if entity.PeerData != nil {
|
||||
hostDataBucket := ctx.Bucket.GetOrCreateBucket(FieldServerPeerData)
|
||||
for k, v := range entity.PeerData {
|
||||
key := make([]byte, 4)
|
||||
binary.LittleEndian.PutUint32(key, k)
|
||||
hostDataBucket.PutValue(key, v)
|
||||
if ctx.ProceedWithSet(FieldServerPeerData) {
|
||||
_ = ctx.Bucket.DeleteBucket([]byte(FieldServerPeerData))
|
||||
if entity.PeerData != nil {
|
||||
hostDataBucket := ctx.Bucket.GetOrCreateBucket(FieldServerPeerData)
|
||||
for k, v := range entity.PeerData {
|
||||
key := make([]byte, 4)
|
||||
binary.LittleEndian.PutUint32(key, k)
|
||||
hostDataBucket.PutValue(key, v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
"fmt"
|
||||
"github.com/google/uuid"
|
||||
"github.com/openziti/fabric/controller/xt"
|
||||
"github.com/openziti/foundation/storage/boltz"
|
||||
"github.com/openziti/foundation/util/stringz"
|
||||
"go.etcd.io/bbolt"
|
||||
"math"
|
||||
@@ -38,6 +39,7 @@ func Test_TerminatorStore(t *testing.T) {
|
||||
t.Run("test create/delete terminators", ctx.testLoadQueryTerminators)
|
||||
t.Run("test update terminators", ctx.testUpdateTerminators)
|
||||
t.Run("test delete terminators", ctx.testDeleteTerminators)
|
||||
t.Run("test patch terminators", ctx.testPatchTerminator)
|
||||
}
|
||||
|
||||
func (ctx *TestContext) testCreateInvalidTerminators(t *testing.T) {
|
||||
@@ -229,6 +231,7 @@ func (ctx *TestContext) testUpdateTerminators(t *testing.T) {
|
||||
|
||||
terminator.Service = e.service.Id // service should not be updated
|
||||
ctx.ValidateUpdated(terminator)
|
||||
|
||||
}
|
||||
|
||||
func (ctx *TestContext) testDeleteTerminators(t *testing.T) {
|
||||
@@ -247,6 +250,41 @@ func (ctx *TestContext) testDeleteTerminators(t *testing.T) {
|
||||
ctx.ValidateDeleted(e.terminator.Id)
|
||||
}
|
||||
|
||||
func (ctx *TestContext) testPatchTerminator(t *testing.T) {
|
||||
service := ctx.requireNewService()
|
||||
router := ctx.requireNewRouter()
|
||||
|
||||
terminator := &Terminator{}
|
||||
terminator.Service = service.Id
|
||||
terminator.Router = router.Id
|
||||
terminator.Binding = uuid.New().String()
|
||||
terminator.Address = uuid.New().String()
|
||||
terminator.Cost = 0
|
||||
terminator.PeerData = map[uint32][]byte{
|
||||
1: {1, 2, 3},
|
||||
2: {3, 4, 5},
|
||||
}
|
||||
ctx.RequireCreate(terminator)
|
||||
ctx.ValidateBaseline(terminator)
|
||||
|
||||
terminator.Cost = 100
|
||||
checker := boltz.MapFieldChecker{
|
||||
FieldTerminatorCost: struct{}{},
|
||||
}
|
||||
ctx.RequirePatch(terminator, checker)
|
||||
ctx.ValidateUpdated(terminator)
|
||||
|
||||
terminator.PeerData = map[uint32][]byte{
|
||||
1: {7, 8, 9},
|
||||
}
|
||||
|
||||
checker = boltz.MapFieldChecker{
|
||||
FieldServerPeerData: struct{}{},
|
||||
}
|
||||
ctx.RequirePatch(terminator, checker)
|
||||
ctx.ValidateUpdated(terminator)
|
||||
}
|
||||
|
||||
type testStrategyFactory struct{}
|
||||
|
||||
func (t testStrategyFactory) GetStrategyName() string {
|
||||
|
||||
@@ -315,7 +315,7 @@ func (buffer *LinkSendBuffer) retransmit() {
|
||||
}
|
||||
|
||||
if retransmitted > 0 {
|
||||
log.Infof("retransmitted [%d] payloads, [%d] buffered, linkSendBufferSize: %d", retransmitted, len(buffer.buffer), buffer.linkSendBufferSize)
|
||||
log.Debugf("retransmitted [%d] payloads, [%d] buffered, linkSendBufferSize: %d", retransmitted, len(buffer.buffer), buffer.linkSendBufferSize)
|
||||
}
|
||||
buffer.lastRetransmitTime = now
|
||||
}
|
||||
|
||||
@@ -421,6 +421,11 @@ func (self *Xgress) tx() {
|
||||
func (self *Xgress) flushSendThenClose() {
|
||||
self.CloseTimeout(self.Options.MaxCloseWait)
|
||||
self.ForwardEndOfSession(func(payload *Payload) bool {
|
||||
if self.payloadBuffer.closed.Get() {
|
||||
// Avoid spurious 'failed to forward payload' error if the buffer is already closed
|
||||
return false
|
||||
}
|
||||
|
||||
pfxlog.ContextLogger(self.Label()).Debug("sending end of session payload")
|
||||
return self.forwardPayload(payload)
|
||||
})
|
||||
@@ -495,7 +500,7 @@ func (self *Xgress) forwardPayload(payload *Payload) bool {
|
||||
sendCallback, err := self.payloadBuffer.BufferPayload(payload)
|
||||
|
||||
if err != nil {
|
||||
pfxlog.ContextLogger(self.Label()).WithError(err).Error("failure forwarding payload")
|
||||
pfxlog.ContextLogger(self.Label()).WithError(err).Error("failure to buffer payload")
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user