Xray-core/proxy/vless/inbound/inbound.go

645 lines
20 KiB
Go
Raw Normal View History

2020-11-25 11:01:53 +00:00
package inbound
2020-12-04 01:36:16 +00:00
//go:generate go run github.com/xtls/xray-core/common/errors/errorgen
2020-11-25 11:01:53 +00:00
import (
"bytes"
2020-11-25 11:01:53 +00:00
"context"
"io"
"reflect"
2020-11-25 11:01:53 +00:00
"strconv"
2021-01-13 15:13:51 +00:00
"strings"
2020-11-25 11:01:53 +00:00
"syscall"
"time"
"unsafe"
2020-11-25 11:01:53 +00:00
"github.com/pires/go-proxyproto"
2020-12-04 01:36:16 +00:00
"github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/errors"
"github.com/xtls/xray-core/common/log"
"github.com/xtls/xray-core/common/net"
"github.com/xtls/xray-core/common/platform"
"github.com/xtls/xray-core/common/protocol"
"github.com/xtls/xray-core/common/retry"
"github.com/xtls/xray-core/common/session"
"github.com/xtls/xray-core/common/signal"
"github.com/xtls/xray-core/common/task"
core "github.com/xtls/xray-core/core"
"github.com/xtls/xray-core/features/dns"
feature_inbound "github.com/xtls/xray-core/features/inbound"
"github.com/xtls/xray-core/features/policy"
"github.com/xtls/xray-core/features/routing"
"github.com/xtls/xray-core/features/stats"
"github.com/xtls/xray-core/proxy/vless"
"github.com/xtls/xray-core/proxy/vless/encoding"
"github.com/xtls/xray-core/transport/internet/stat"
2020-12-04 01:36:16 +00:00
"github.com/xtls/xray-core/transport/internet/tls"
"github.com/xtls/xray-core/transport/internet/xtls"
2020-11-25 11:01:53 +00:00
)
2021-10-19 16:57:14 +00:00
var xtls_show = false
2020-11-25 11:01:53 +00:00
func init() {
common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, config interface{}) (interface{}, error) {
var dc dns.Client
if err := core.RequireFeatures(ctx, func(d dns.Client) error {
dc = d
return nil
}); err != nil {
return nil, err
}
return New(ctx, config.(*Config), dc)
}))
const defaultFlagValue = "NOT_DEFINED_AT_ALL"
xtlsShow := platform.NewEnvFlag("xray.vless.xtls.show").GetValue(func() string { return defaultFlagValue })
if xtlsShow == "true" {
xtls_show = true
}
}
// Handler is an inbound connection handler that handles messages in VLess protocol.
type Handler struct {
inboundHandlerManager feature_inbound.Manager
policyManager policy.Manager
validator *vless.Validator
dns dns.Client
2021-01-13 15:13:51 +00:00
fallbacks map[string]map[string]map[string]*Fallback // or nil
2020-11-25 11:01:53 +00:00
// regexps map[string]*regexp.Regexp // or nil
}
// New creates a new VLess inbound handler.
func New(ctx context.Context, config *Config, dc dns.Client) (*Handler, error) {
v := core.MustFromContext(ctx)
handler := &Handler{
inboundHandlerManager: v.GetFeature(feature_inbound.ManagerType()).(feature_inbound.Manager),
policyManager: v.GetFeature(policy.ManagerType()).(policy.Manager),
validator: new(vless.Validator),
dns: dc,
}
for _, user := range config.Clients {
u, err := user.ToMemoryUser()
if err != nil {
return nil, newError("failed to get VLESS user").Base(err).AtError()
}
if err := handler.AddUser(ctx, u); err != nil {
return nil, newError("failed to initiate user").Base(err).AtError()
}
}
if config.Fallbacks != nil {
2021-01-13 15:13:51 +00:00
handler.fallbacks = make(map[string]map[string]map[string]*Fallback)
2020-11-25 11:01:53 +00:00
// handler.regexps = make(map[string]*regexp.Regexp)
for _, fb := range config.Fallbacks {
2021-01-13 15:13:51 +00:00
if handler.fallbacks[fb.Name] == nil {
handler.fallbacks[fb.Name] = make(map[string]map[string]*Fallback)
2020-11-25 11:01:53 +00:00
}
2021-01-13 15:13:51 +00:00
if handler.fallbacks[fb.Name][fb.Alpn] == nil {
handler.fallbacks[fb.Name][fb.Alpn] = make(map[string]*Fallback)
}
handler.fallbacks[fb.Name][fb.Alpn][fb.Path] = fb
2020-11-25 11:01:53 +00:00
/*
if fb.Path != "" {
if r, err := regexp.Compile(fb.Path); err != nil {
return nil, newError("invalid path regexp").Base(err).AtError()
} else {
handler.regexps[fb.Path] = r
}
}
*/
}
if handler.fallbacks[""] != nil {
for name, apfb := range handler.fallbacks {
if name != "" {
for alpn := range handler.fallbacks[""] {
if apfb[alpn] == nil {
apfb[alpn] = make(map[string]*Fallback)
}
}
}
}
}
2021-01-14 21:55:52 +00:00
for _, apfb := range handler.fallbacks {
if apfb[""] != nil {
for alpn, pfb := range apfb {
if alpn != "" { // && alpn != "h2" {
for path, fb := range apfb[""] {
if pfb[path] == nil {
pfb[path] = fb
}
}
}
}
}
}
2020-11-25 11:01:53 +00:00
if handler.fallbacks[""] != nil {
2021-01-14 21:55:52 +00:00
for name, apfb := range handler.fallbacks {
if name != "" {
for alpn, pfb := range handler.fallbacks[""] {
for path, fb := range pfb {
if apfb[alpn][path] == nil {
apfb[alpn][path] = fb
}
2020-11-25 11:01:53 +00:00
}
}
}
}
}
}
return handler, nil
}
// Close implements common.Closable.Close().
func (h *Handler) Close() error {
return errors.Combine(common.Close(h.validator))
}
// AddUser implements proxy.UserManager.AddUser().
func (h *Handler) AddUser(ctx context.Context, u *protocol.MemoryUser) error {
return h.validator.Add(u)
}
// RemoveUser implements proxy.UserManager.RemoveUser().
func (h *Handler) RemoveUser(ctx context.Context, e string) error {
return h.validator.Del(e)
}
// Network implements proxy.Inbound.Network().
func (*Handler) Network() []net.Network {
return []net.Network{net.Network_TCP, net.Network_UNIX}
}
// Process implements proxy.Inbound.Process().
func (h *Handler) Process(ctx context.Context, network net.Network, connection stat.Connection, dispatcher routing.Dispatcher) error {
2020-11-25 11:01:53 +00:00
sid := session.ExportIDToError(ctx)
iConn := connection
statConn, ok := iConn.(*stat.CounterConnection)
2020-11-25 11:01:53 +00:00
if ok {
iConn = statConn.Connection
}
sessionPolicy := h.policyManager.ForLevel(0)
if err := connection.SetReadDeadline(time.Now().Add(sessionPolicy.Timeouts.Handshake)); err != nil {
return newError("unable to set read deadline").Base(err).AtWarning()
}
first := buf.FromBytes(make([]byte, buf.Size))
first.Clear()
2020-11-25 11:01:53 +00:00
firstLen, _ := first.ReadFrom(connection)
newError("firstLen = ", firstLen).AtInfo().WriteToLog(sid)
reader := &buf.BufferedReader{
Reader: buf.NewReader(connection),
Buffer: buf.MultiBuffer{first},
}
var request *protocol.RequestHeader
var requestAddons *encoding.Addons
var err error
2021-01-14 21:55:52 +00:00
napfb := h.fallbacks
isfb := napfb != nil
2020-11-25 11:01:53 +00:00
if isfb && firstLen < 18 {
err = newError("fallback directly")
} else {
request, requestAddons, isfb, err = encoding.DecodeRequestHeader(isfb, first, reader, h.validator)
}
if err != nil {
if isfb {
if err := connection.SetReadDeadline(time.Time{}); err != nil {
newError("unable to set back read deadline").Base(err).AtWarning().WriteToLog(sid)
}
newError("fallback starts").Base(err).AtInfo().WriteToLog(sid)
2021-01-13 15:13:51 +00:00
name := ""
2020-11-25 11:01:53 +00:00
alpn := ""
2021-01-14 21:55:52 +00:00
if tlsConn, ok := iConn.(*tls.Conn); ok {
cs := tlsConn.ConnectionState()
name = cs.ServerName
alpn = cs.NegotiatedProtocol
newError("realName = " + name).AtInfo().WriteToLog(sid)
newError("realAlpn = " + alpn).AtInfo().WriteToLog(sid)
} else if xtlsConn, ok := iConn.(*xtls.Conn); ok {
cs := xtlsConn.ConnectionState()
name = cs.ServerName
alpn = cs.NegotiatedProtocol
newError("realName = " + name).AtInfo().WriteToLog(sid)
newError("realAlpn = " + alpn).AtInfo().WriteToLog(sid)
}
name = strings.ToLower(name)
alpn = strings.ToLower(alpn)
2021-01-14 21:55:52 +00:00
if len(napfb) > 1 || napfb[""] == nil {
if name != "" && napfb[name] == nil {
match := ""
for n := range napfb {
if n != "" && strings.Contains(name, n) && len(n) > len(match) {
match = n
}
2021-01-13 15:13:51 +00:00
}
name = match
2021-01-13 15:13:51 +00:00
}
2021-01-14 21:55:52 +00:00
}
if napfb[name] == nil {
name = ""
}
apfb := napfb[name]
if apfb == nil {
return newError(`failed to find the default "name" config`).AtWarning()
}
2021-01-13 15:13:51 +00:00
2021-01-14 21:55:52 +00:00
if apfb[alpn] == nil {
alpn = ""
2020-11-25 11:01:53 +00:00
}
2021-01-14 21:55:52 +00:00
pfb := apfb[alpn]
2020-11-25 11:01:53 +00:00
if pfb == nil {
return newError(`failed to find the default "alpn" config`).AtWarning()
}
path := ""
if len(pfb) > 1 || pfb[""] == nil {
/*
if lines := bytes.Split(firstBytes, []byte{'\r', '\n'}); len(lines) > 1 {
if s := bytes.Split(lines[0], []byte{' '}); len(s) == 3 {
if len(s[0]) < 8 && len(s[1]) > 0 && len(s[2]) == 8 {
newError("realPath = " + string(s[1])).AtInfo().WriteToLog(sid)
for _, fb := range pfb {
if fb.Path != "" && h.regexps[fb.Path].Match(s[1]) {
path = fb.Path
break
}
}
}
}
}
*/
if firstLen >= 18 && first.Byte(4) != '*' { // not h2c
firstBytes := first.Bytes()
for i := 4; i <= 8; i++ { // 5 -> 9
if firstBytes[i] == '/' && firstBytes[i-1] == ' ' {
search := len(firstBytes)
if search > 64 {
search = 64 // up to about 60
}
for j := i + 1; j < search; j++ {
k := firstBytes[j]
if k == '\r' || k == '\n' { // avoid logging \r or \n
break
}
if k == '?' || k == ' ' {
2020-11-25 11:01:53 +00:00
path = string(firstBytes[i:j])
newError("realPath = " + path).AtInfo().WriteToLog(sid)
if pfb[path] == nil {
path = ""
}
break
}
}
break
}
}
}
}
fb := pfb[path]
if fb == nil {
return newError(`failed to find the default "path" config`).AtWarning()
}
ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
ctx = policy.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
var conn net.Conn
if err := retry.ExponentialBackoff(5, 100).On(func() error {
var dialer net.Dialer
conn, err = dialer.DialContext(ctx, fb.Type, fb.Dest)
if err != nil {
return err
}
return nil
}); err != nil {
return newError("failed to dial to " + fb.Dest).Base(err).AtWarning()
}
defer conn.Close()
serverReader := buf.NewReader(conn)
serverWriter := buf.NewWriter(conn)
postRequest := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
if fb.Xver != 0 {
2021-01-23 21:06:15 +00:00
ipType := 4
remoteAddr, remotePort, err := net.SplitHostPort(connection.RemoteAddr().String())
if err != nil {
ipType = 0
}
localAddr, localPort, err := net.SplitHostPort(connection.LocalAddr().String())
if err != nil {
ipType = 0
}
if ipType == 4 {
for i := 0; i < len(remoteAddr); i++ {
if remoteAddr[i] == ':' {
ipType = 6
break
}
2020-11-25 11:01:53 +00:00
}
}
pro := buf.New()
defer pro.Release()
switch fb.Xver {
case 1:
if ipType == 0 {
pro.Write([]byte("PROXY UNKNOWN\r\n"))
break
}
if ipType == 4 {
2020-11-25 11:01:53 +00:00
pro.Write([]byte("PROXY TCP4 " + remoteAddr + " " + localAddr + " " + remotePort + " " + localPort + "\r\n"))
} else {
pro.Write([]byte("PROXY TCP6 " + remoteAddr + " " + localAddr + " " + remotePort + " " + localPort + "\r\n"))
}
case 2:
pro.Write([]byte("\x0D\x0A\x0D\x0A\x00\x0D\x0A\x51\x55\x49\x54\x0A")) // signature
if ipType == 0 {
pro.Write([]byte("\x20\x00\x00\x00")) // v2 + LOCAL + UNSPEC + UNSPEC + 0 bytes
break
}
if ipType == 4 {
pro.Write([]byte("\x21\x11\x00\x0C")) // v2 + PROXY + AF_INET + STREAM + 12 bytes
2020-11-25 11:01:53 +00:00
pro.Write(net.ParseIP(remoteAddr).To4())
pro.Write(net.ParseIP(localAddr).To4())
} else {
pro.Write([]byte("\x21\x21\x00\x24")) // v2 + PROXY + AF_INET6 + STREAM + 36 bytes
2020-11-25 11:01:53 +00:00
pro.Write(net.ParseIP(remoteAddr).To16())
pro.Write(net.ParseIP(localAddr).To16())
}
p1, _ := strconv.ParseUint(remotePort, 10, 16)
p2, _ := strconv.ParseUint(localPort, 10, 16)
pro.Write([]byte{byte(p1 >> 8), byte(p1), byte(p2 >> 8), byte(p2)})
}
if err := serverWriter.WriteMultiBuffer(buf.MultiBuffer{pro}); err != nil {
return newError("failed to set PROXY protocol v", fb.Xver).Base(err).AtWarning()
}
}
if err := buf.Copy(reader, serverWriter, buf.UpdateActivity(timer)); err != nil {
return newError("failed to fallback request payload").Base(err).AtInfo()
}
return nil
}
writer := buf.NewWriter(connection)
getResponse := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
if err := buf.Copy(serverReader, writer, buf.UpdateActivity(timer)); err != nil {
return newError("failed to deliver response payload").Base(err).AtInfo()
}
return nil
}
if err := task.Run(ctx, task.OnSuccess(postRequest, task.Close(serverWriter)), task.OnSuccess(getResponse, task.Close(writer))); err != nil {
common.Interrupt(serverReader)
common.Interrupt(serverWriter)
return newError("fallback ends").Base(err).AtInfo()
}
return nil
}
if errors.Cause(err) != io.EOF {
log.Record(&log.AccessMessage{
From: connection.RemoteAddr(),
To: "",
Status: log.AccessRejected,
Reason: err,
})
err = newError("invalid request from ", connection.RemoteAddr()).Base(err).AtInfo()
}
return err
}
if err := connection.SetReadDeadline(time.Time{}); err != nil {
newError("unable to set back read deadline").Base(err).AtWarning().WriteToLog(sid)
}
newError("received request for ", request.Destination()).AtInfo().WriteToLog(sid)
inbound := session.InboundFromContext(ctx)
if inbound == nil {
panic("no inbound metadata")
}
inbound.User = request.User
account := request.User.Account.(*vless.MemoryAccount)
responseAddons := &encoding.Addons{
// Flow: requestAddons.Flow,
}
var netConn net.Conn
2020-11-25 11:01:53 +00:00
var rawConn syscall.RawConn
var input *bytes.Reader
var rawInput *bytes.Buffer
allowNoneFlow := false
accountFlow := account.Flow
flows := strings.Split(account.Flow, ",")
for _, f := range flows {
t := strings.TrimSpace(f)
if t == "none" {
allowNoneFlow = true
} else {
accountFlow = t
}
}
2020-11-25 11:01:53 +00:00
switch requestAddons.Flow {
case vless.XRO, vless.XRD, vless.XRV:
if accountFlow == requestAddons.Flow {
2020-11-25 11:01:53 +00:00
switch request.Command {
case protocol.RequestCommandMux:
return newError(requestAddons.Flow + " doesn't support Mux").AtWarning()
case protocol.RequestCommandUDP:
return newError(requestAddons.Flow + " doesn't support UDP").AtWarning()
case protocol.RequestCommandTCP:
if requestAddons.Flow == vless.XRV {
var t reflect.Type
var p uintptr
if tlsConn, ok := iConn.(*tls.Conn); ok {
netConn = tlsConn.NetConn()
if pc, ok := netConn.(*proxyproto.Conn); ok {
netConn = pc.Raw()
// 8192 > 4096, there is no need to process pc's bufReader
}
2022-11-13 17:18:23 +00:00
if sc, ok := netConn.(syscall.Conn); ok {
rawConn, _ = sc.SyscallConn()
}
t = reflect.TypeOf(tlsConn.Conn).Elem()
p = uintptr(unsafe.Pointer(tlsConn.Conn))
} else if _, ok := iConn.(*tls.UConn); ok {
return newError("XTLS only supports UTLS fingerprint for the outbound.").AtWarning()
} else if _, ok := iConn.(*xtls.Conn); ok {
return newError(`failed to use ` + requestAddons.Flow + `, vision "security" must be "tls"`).AtWarning()
} else {
return newError("XTLS only supports TCP, mKCP and DomainSocket for now.").AtWarning()
}
i, _ := t.FieldByName("input")
r, _ := t.FieldByName("rawInput")
input = (*bytes.Reader)(unsafe.Pointer(p + i.Offset))
rawInput = (*bytes.Buffer)(unsafe.Pointer(p + r.Offset))
} else if xtlsConn, ok := iConn.(*xtls.Conn); ok {
2020-11-25 11:01:53 +00:00
xtlsConn.RPRX = true
xtlsConn.SHOW = xtls_show
xtlsConn.MARK = "XTLS"
if requestAddons.Flow == vless.XRD {
xtlsConn.DirectMode = true
2022-09-11 04:20:06 +00:00
if sc, ok := xtlsConn.NetConn().(syscall.Conn); ok {
2020-11-25 11:01:53 +00:00
rawConn, _ = sc.SyscallConn()
}
}
} else {
return newError(`failed to use ` + requestAddons.Flow + `, maybe "security" is not "xtls"`).AtWarning()
}
}
} else {
return newError(account.ID.String() + " is not able to use " + requestAddons.Flow).AtWarning()
}
case "", "none":
if accountFlow == vless.XRV && !allowNoneFlow && request.Command == protocol.RequestCommandTCP {
2022-12-26 00:37:35 +00:00
return newError(account.ID.String() + " is not able to use " + vless.XRV +
". Note the pure tls proxy has certain tls in tls characters. Append \",none\" in flow to suppress").AtWarning()
}
2020-11-25 11:01:53 +00:00
default:
return newError("unknown request flow " + requestAddons.Flow).AtWarning()
}
if request.Command != protocol.RequestCommandMux {
ctx = log.ContextWithAccessMessage(ctx, &log.AccessMessage{
From: connection.RemoteAddr(),
To: request.Destination(),
Status: log.AccessAccepted,
Reason: "",
Email: request.User.Email,
})
}
sessionPolicy = h.policyManager.ForLevel(request.User.Level)
ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
ctx = policy.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
link, err := dispatcher.Dispatch(ctx, request.Destination())
if err != nil {
return newError("failed to dispatch request to ", request.Destination()).Base(err).AtWarning()
}
serverReader := link.Reader // .(*pipe.Reader)
serverWriter := link.Writer // .(*pipe.Writer)
2022-11-13 17:18:23 +00:00
enableXtls := false
isTLS12orAbove := false
isTLS := false
2022-11-22 03:37:22 +00:00
var cipher uint16 = 0
var remainingServerHello int32 = -1
numberOfPacketToFilter := 8
2020-11-25 11:01:53 +00:00
postRequest := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.DownlinkOnly)
// default: clientReader := reader
clientReader := encoding.DecodeBodyAddons(reader, request, requestAddons)
var err error
if rawConn != nil {
var counter stats.Counter
if statConn != nil {
counter = statConn.ReadCounter
}
2022-12-26 00:37:35 +00:00
// TODO enable splice
2022-11-07 02:50:19 +00:00
ctx = session.ContextWithInbound(ctx, nil)
if requestAddons.Flow == vless.XRV {
err = encoding.XtlsRead(clientReader, serverWriter, timer, netConn, rawConn, input, rawInput, counter, ctx, account.ID.Bytes(),
2022-12-26 00:37:35 +00:00
&numberOfPacketToFilter, &enableXtls, &isTLS12orAbove, &isTLS, &cipher, &remainingServerHello)
} else {
err = encoding.ReadV(clientReader, serverWriter, timer, iConn.(*xtls.Conn), rawConn, counter, ctx)
}
2020-11-25 11:01:53 +00:00
} else {
// from clientReader.ReadMultiBuffer to serverWriter.WriteMultiBufer
err = buf.Copy(clientReader, serverWriter, buf.UpdateActivity(timer))
}
if err != nil {
return newError("failed to transfer request payload").Base(err).AtInfo()
}
return nil
}
getResponse := func() error {
defer timer.SetTimeout(sessionPolicy.Timeouts.UplinkOnly)
bufferWriter := buf.NewBufferedWriter(buf.NewWriter(connection))
if err := encoding.EncodeResponseHeader(bufferWriter, request, responseAddons); err != nil {
return newError("failed to encode response header").Base(err).AtWarning()
}
// default: clientWriter := bufferWriter
clientWriter := encoding.EncodeBodyAddons(bufferWriter, request, responseAddons)
userUUID := account.ID.Bytes()
multiBuffer, err1 := serverReader.ReadMultiBuffer()
if err1 != nil {
return err1 // ...
}
if requestAddons.Flow == vless.XRV {
2022-11-22 03:37:22 +00:00
encoding.XtlsFilterTls(multiBuffer, &numberOfPacketToFilter, &enableXtls, &isTLS12orAbove, &isTLS, &cipher, &remainingServerHello, ctx)
if isTLS {
multiBuffer = encoding.ReshapeMultiBuffer(ctx, multiBuffer)
for i, b := range multiBuffer {
multiBuffer[i] = encoding.XtlsPadding(b, 0x00, &userUUID, ctx)
}
2020-11-25 11:01:53 +00:00
}
}
if err := clientWriter.WriteMultiBuffer(multiBuffer); err != nil {
return err // ...
}
2020-11-25 11:01:53 +00:00
// Flush; bufferWriter.WriteMultiBufer now is bufferWriter.writer.WriteMultiBuffer
if err := bufferWriter.SetBuffered(false); err != nil {
return newError("failed to write A response payload").Base(err).AtWarning()
}
var err error
if rawConn != nil && requestAddons.Flow == vless.XRV {
var counter stats.Counter
if statConn != nil {
counter = statConn.WriteCounter
}
2022-12-26 00:37:35 +00:00
err = encoding.XtlsWrite(serverReader, clientWriter, timer, netConn, counter, ctx, &userUUID, &numberOfPacketToFilter,
2022-11-22 03:37:22 +00:00
&enableXtls, &isTLS12orAbove, &isTLS, &cipher, &remainingServerHello)
} else {
// from serverReader.ReadMultiBuffer to clientWriter.WriteMultiBufer
err = buf.Copy(serverReader, clientWriter, buf.UpdateActivity(timer))
}
if err != nil {
2020-11-25 11:01:53 +00:00
return newError("failed to transfer response payload").Base(err).AtInfo()
}
// Indicates the end of response payload.
switch responseAddons.Flow {
default:
}
return nil
}
if err := task.Run(ctx, task.OnSuccess(postRequest, task.Close(serverWriter)), getResponse); err != nil {
common.Interrupt(serverReader)
common.Interrupt(serverWriter)
return newError("connection ends").Base(err).AtInfo()
}
return nil
}