| 152 | } |
| 153 | |
| 154 | func SendData(stream remote.Tunnel_InitTunnelClient, sessions <-chan *tunnel.Session, closeChan <-chan bool, log logpkg.Logger) error { |
| 155 | for { |
| 156 | select { |
| 157 | case <-stream.Context().Done(): |
| 158 | return nil |
| 159 | case <-closeChan: |
| 160 | return nil |
| 161 | case session := <-sessions: |
| 162 | // read the bytes from the buffer |
| 163 | // but allow it to keep growing while we send the response |
| 164 | session.Lock() |
| 165 | bys := session.Buf.Len() |
| 166 | bytes := make([]byte, bys) |
| 167 | _, err := session.Buf.Read(bytes) |
| 168 | if err != nil { |
| 169 | session.Unlock() |
| 170 | return fmt.Errorf("failed reading stream from session %v, exiting", err) |
| 171 | } |
| 172 | log.Debugf("read %d from buffer out of %d available", len(bytes), bys) |
| 173 | resp := &remote.SocketDataRequest{ |
| 174 | RequestId: session.ID.String(), |
| 175 | Data: bytes, |
| 176 | ShouldClose: !session.Open, |
| 177 | } |
| 178 | session.Unlock() |
| 179 | |
| 180 | log.Debugf("sending %d bytes to server", len(bytes)) |
| 181 | err = stream.Send(resp) |
| 182 | if err != nil { |
| 183 | return fmt.Errorf("failed sending message to tunnel stream, exiting") |
| 184 | } |
| 185 | log.Debugf("sent %d bytes to server", len(bytes)) |
| 186 | } |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | func StartReverseForward(ctx context.Context, reader io.ReadCloser, writer io.WriteCloser, tunnels []*latest.PortMapping, stopChan chan struct{}, namespace string, name string, log logpkg.Logger) error { |
| 191 | scheme := "TCP" |