starred/godoxy
1
0
Fork 0
mirror of https://github.com/yusing/godoxy.git synced 2026-04-24 17:58:45 +02:00
Code Issues 18 Packages Projects Releases 10 Wiki Activity

modules reorganized and code refactor

Browse Source
This commit is contained in:
yusing
2024-11-25 01:40:12 +08:00
parent f3b21e6bd9
commit d723403b6b
46 changed files with 437 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
internal/route/stream.go
View File

@@ -9,9 +9,9 @@ import (
"github.com/yusing/go-proxy/internal/docker/idlewatcher"
E "github.com/yusing/go-proxy/internal/error"
net "github.com/yusing/go-proxy/internal/net/types"
"github.com/yusing/go-proxy/internal/proxy/entry"
"github.com/yusing/go-proxy/internal/route/entry"
"github.com/yusing/go-proxy/internal/route/routes"
"github.com/yusing/go-proxy/internal/task"
F "github.com/yusing/go-proxy/internal/utils/functional"
"github.com/yusing/go-proxy/internal/watcher/health"
"github.com/yusing/go-proxy/internal/watcher/health/monitor"
)
@@ -19,7 +19,7 @@ import (
type StreamRoute struct {
*entry.StreamEntry
stream net.Stream
net.Stream
HealthMon health.HealthMonitor `json:"health"`
@@ -28,12 +28,6 @@ type StreamRoute struct {
l zerolog.Logger
}
var streamRoutes = F.NewMapOf[string, *StreamRoute]()
func GetStreamProxies() F.Map[string, *StreamRoute] {
return streamRoutes
}
func NewStreamRoute(entry *entry.StreamEntry) (impl, E.Error) {
// TODO: support non-coherent scheme
if !entry.Scheme.IsCoherent() {
@@ -60,29 +54,29 @@ func (r *StreamRoute) Start(providerSubtask task.Task) E.Error {
}
r.task = providerSubtask
r.stream = NewStream(r)
r.Stream = NewStream(r)
switch {
case entry.UseIdleWatcher(r):
wakerTask := providerSubtask.Parent().Subtask("waker for " + string(r.Alias))
waker, err := idlewatcher.NewStreamWaker(wakerTask, r.StreamEntry, r.stream)
waker, err := idlewatcher.NewStreamWaker(wakerTask, r.StreamEntry, r.Stream)
if err != nil {
r.task.Finish(err)
return err
}
r.stream = waker
r.Stream = waker
r.HealthMon = waker
case entry.UseHealthCheck(r):
r.HealthMon = monitor.NewRawHealthMonitor(r.TargetURL(), r.HealthCheck)
}
if err := r.stream.Setup(); err != nil {
if err := r.Stream.Setup(); err != nil {
r.task.Finish(err)
return E.From(err)
}
r.task.OnFinished("close stream", func() {
if err := r.stream.Close(); err != nil {
if err := r.Stream.Close(); err != nil {
E.LogError("close stream failed", err, &r.l)
}
})
@@ -101,9 +95,9 @@ func (r *StreamRoute) Start(providerSubtask task.Task) E.Error {
go r.acceptConnections()
streamRoutes.Store(string(r.Alias), r)
routes.SetStreamRoute(string(r.Alias), r)
r.task.OnFinished("remove from route table", func() {
streamRoutes.Delete(string(r.Alias))
routes.DeleteStreamRoute(string(r.Alias))
})
return nil
}
@@ -120,7 +114,7 @@ func (r *StreamRoute) acceptConnections() {
case <-r.task.Context().Done():
return
default:
conn, err := r.stream.Accept()
conn, err := r.Stream.Accept()
if err != nil {
select {
case <-r.task.Context().Done():
@@ -135,7 +129,7 @@ func (r *StreamRoute) acceptConnections() {
}
connTask := r.task.Subtask("connection")
go func() {
err := r.stream.Handle(conn)
err := r.Stream.Handle(conn)
if err != nil && !errors.Is(err, context.Canceled) {
E.LogError("handle connection error", err, &r.l)
connTask.Finish(err)