| 53 | } |
| 54 | |
| 55 | func SendData(stream remote.Tunnel_InitTunnelServer, sessions <-chan *Session, closeChan chan struct{}) { |
| 56 | for { |
| 57 | select { |
| 58 | case <-closeChan: |
| 59 | return |
| 60 | case <-stream.Context().Done(): |
| 61 | return |
| 62 | case session := <-sessions: |
| 63 | // read the bytes from the buffer |
| 64 | // but allow it to keep growing while we send the response |
| 65 | session.Lock() |
| 66 | bys := session.Buf.Len() |
| 67 | bytes := make([]byte, bys) |
| 68 | _, _ = session.Buf.Read(bytes) |
| 69 | resp := &remote.SocketDataResponse{ |
| 70 | HasErr: false, |
| 71 | LogMessage: nil, |
| 72 | Data: bytes, |
| 73 | RequestId: session.ID.String(), |
| 74 | ShouldClose: !session.Open, |
| 75 | } |
| 76 | session.Unlock() |
| 77 | |
| 78 | stderrlog.Debugf("sending %d bytes to client", len(bytes)) |
| 79 | err := stream.Send(resp) |
| 80 | if err != nil { |
| 81 | stderrlog.Errorf("failed sending message to tunnel stream: %v", err) |
| 82 | close(closeChan) |
| 83 | return |
| 84 | } |
| 85 | stderrlog.Debugf("sent %d bytes to client", len(bytes)) |
| 86 | } |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | func ReceiveData(stream remote.Tunnel_InitTunnelServer, closeChan chan struct{}) { |
| 91 | for { |