(stream remote.Tunnel_InitTunnelClient, closeStream <-chan bool, sessionsOut chan<- *tunnel.Session, port int32, scheme string, log logpkg.Logger)
| 20 | ) |
| 21 | |
| 22 | func ReceiveData(stream remote.Tunnel_InitTunnelClient, closeStream <-chan bool, sessionsOut chan<- *tunnel.Session, port int32, scheme string, log logpkg.Logger) error { |
| 23 | loop: |
| 24 | for { |
| 25 | m, err := stream.Recv() |
| 26 | select { |
| 27 | case <-closeStream: |
| 28 | log.Debugf("closing listener on %d", port) |
| 29 | _ = stream.CloseSend() |
| 30 | break loop |
| 31 | case <-stream.Context().Done(): |
| 32 | _ = stream.CloseSend() |
| 33 | break loop |
| 34 | default: |
| 35 | if err != nil { |
| 36 | return fmt.Errorf("error reading from stream: %v", err) |
| 37 | } else if m.HasErr { |
| 38 | _ = stream.CloseSend() |
| 39 | if m.LogMessage == nil { |
| 40 | return fmt.Errorf("remote error: unknown") |
| 41 | } |
| 42 | |
| 43 | return fmt.Errorf("helper error: %s", m.LogMessage.Message) |
| 44 | } |
| 45 | |
| 46 | requestID, err := uuid.Parse(m.RequestId) |
| 47 | if err != nil { |
| 48 | log.Errorf("%s; failed parsing session uuid from stream, skipping", m.RequestId) |
| 49 | continue |
| 50 | } |
| 51 | |
| 52 | session, exists := tunnel.GetSession(requestID) |
| 53 | if !exists { |
| 54 | log.Debugf("new connection %s", requestID) |
| 55 | |
| 56 | // new session |
| 57 | conn, err := net.DialTimeout(strings.ToLower(scheme), fmt.Sprintf("localhost:%d", port), time.Millisecond*500) |
| 58 | if err != nil { |
| 59 | log.Errorf("failed connecting to localhost on port %d scheme %s: %v", port, scheme, err) |
| 60 | // close the remote connection |
| 61 | resp := &remote.SocketDataRequest{ |
| 62 | RequestId: requestID.String(), |
| 63 | ShouldClose: true, |
| 64 | } |
| 65 | err := stream.Send(resp) |
| 66 | if err != nil { |
| 67 | log.Errorf("failed sending close message to tunnel stream: %v", err) |
| 68 | } |
| 69 | |
| 70 | continue |
| 71 | } |
| 72 | |
| 73 | session, err = tunnel.NewSessionFromStream(requestID, conn) |
| 74 | if err != nil { |
| 75 | log.Errorf("%s; error creating new session from stream: %v", m.RequestId, err) |
| 76 | continue |
| 77 | } |
| 78 | |
| 79 | go ReadFromSession(session, sessionsOut, log) |
no test coverage detected