add http connect proxy server

1 files changed, 6 insertions, 5 deletions

diff --git a/pkg/server/tunnel.go b/pkg/server/tunnel.go
index 4543714..e47215e 100644
--- a/pkg/server/tunnel.go
+++ b/pkg/server/tunnel.go
@@ -33,7 +33,7 @@ type stream struct {
 	since   time.Time
 	until   time.Time
 	wg      sync.WaitGroup
-	in, out socket.Channel
+	in, out socket.Conn
 	pipes   []*hook.Pipe
 
 	m struct {
@@ -135,6 +135,8 @@ func (t *tunnel) serve() {
 
 		env := t.env.Fork()
 
+		env.Set("tunnel", t.id)
+
 		if in, err := t.in.Open(env); err == nil {
 			if out, err := t.out.Open(env); err == nil {
 				s := t.newStream(env, in, out)
@@ -157,7 +159,7 @@ func (t *tunnel) serve() {
 	close(t.done)
 }
 
-func (t *tunnel) newStream(env env.Env, in, out socket.Channel) *stream {
+func (t *tunnel) newStream(env env.Env, in, out socket.Conn) *stream {
 	s := &stream{
 		t:     t,
 		in:    in,
@@ -167,7 +169,6 @@ func (t *tunnel) newStream(env env.Env, in, out socket.Channel) *stream {
 		since: time.Now(),
 	}
 
-	s.env.Set("tunnel", t.id)
 	s.env.Set("stream", strconv.Itoa(s.id))
 
 	s.run()
@@ -230,7 +231,7 @@ func (s *stream) waitAndClose() {
 	log.Println(s.t, s, "done", s.info())
 }
 
-func (s *stream) channel(c socket.Channel, m *metric, rq, wq queue.Q) {
+func (s *stream) channel(c socket.Conn, m *metric, rq, wq queue.Q) {
 	watch := func(q queue.Q, f func(q queue.Q) error) {
 		defer s.wg.Done()
 
@@ -249,7 +250,7 @@ func (s *stream) channel(c socket.Channel, m *metric, rq, wq queue.Q) {
 		}
 
 		if err != nil {
-			log.Println(s.t, s, err)
+			log.Println(s.t, s, c, err)
 		}
 	}