web addon content refresh when change

proxy/flow.go

@@ -100,10 +100,6 @@ type Response struct {
 	BodyReader io.Reader
 
 	close bool // connection close
-
-	decodedBody []byte
-	decoded     bool // decoded reports whether the response was sent compressed but was decoded to decodedBody.
-	decodedErr  error
 }
 
 // flow

proxy/flowencoding.go

@@ -36,44 +36,27 @@ func (r *Response) IsTextContentType() bool {
 }
 
 func (r *Response) DecodedBody() ([]byte, error) {
-	if r.decodedBody != nil {
-		return r.decodedBody, nil
-	}
-
-	if r.decodedErr != nil {
-		return nil, r.decodedErr
-	}
-
-	if r.Body == nil {
-		return nil, nil
-	}
-
 	if len(r.Body) == 0 {
-		r.decodedBody = r.Body
-		return r.decodedBody, nil
+		return r.Body, nil
 	}
 
 	enc := r.Header.Get("Content-Encoding")
 	if enc == "" || enc == "identity" {
-		r.decodedBody = r.Body
-		return r.decodedBody, nil
+		return r.Body, nil
 	}
 
 	decodedBody, decodedErr := decode(enc, r.Body)
 	if decodedErr != nil {
-		r.decodedErr = decodedErr
-		log.Error(r.decodedErr)
+		log.Error(decodedErr)
 		return nil, decodedErr
 	}
 
-	r.decodedBody = decodedBody
-	r.decoded = true
-	return r.decodedBody, nil
+	return decodedBody, nil
 }
 
 func (r *Response) ReplaceToDecodedBody() {
 	body, err := r.DecodedBody()
-	if err != nil || body == nil {
+	if err != nil {
 		return
 	}
 

web/client/src/App.tsx

@@ -127,19 +127,26 @@ class App extends React.Component<IProps, IState> {
         this.connMgr.delete(msg.id)
       }
       else if (msg.type === MessageType.REQUEST) {
-        const flow = new Flow(msg, this.connMgr)
-        flow.getConn()
-        this.flowMgr.add(flow)
-
-        let shouldScroll = false
-        if (this.tableBottomRef?.current && isInViewPort(this.tableBottomRef.current)) {
-          shouldScroll = true
-        }
-        this.setState({ flows: this.flowMgr.showList() }, () => {
-          if (shouldScroll) {
-            this.tableBottomRef?.current?.scrollIntoView({ behavior: 'auto' })
+        let flow = this.flowMgr.get(msg.id)
+        if (!flow) {
+          flow = new Flow(msg, this.connMgr)
+          flow.getConn()
+          this.flowMgr.add(flow)
+
+          let shouldScroll = false
+          if (this.tableBottomRef?.current && isInViewPort(this.tableBottomRef.current)) {
+            shouldScroll = true
           }
-        })
+          this.setState({ flows: this.flowMgr.showList() }, () => {
+            if (shouldScroll) {
+              this.tableBottomRef?.current?.scrollIntoView({ behavior: 'auto' })
+            }
+          })
+        } else {
+          flow.addRequest(msg)
+          flow.getConn()
+          this.setState({ flows: this.state.flows })
+        }
       }
       else if (msg.type === MessageType.REQUEST_BODY) {
         const flow = this.flowMgr.get(msg.id)

web/client/src/lib/flow.ts

@@ -46,13 +46,13 @@ export interface IFlowPreview {
 export class Flow {
   public no: number
   public id: string
-  public connId: string
-  public waitIntercept: boolean
-  public request: IRequest
+  public connId!: string
+  public waitIntercept!: boolean
+  public request!: IRequest
   public response: IResponse | null = null
 
-  public url: URL
-  private path: string
+  public url!: URL
+  private path!: string
   private _size = 0
   private size = '0'
   private headerContentLengthExist = false
@@ -82,6 +82,19 @@ export class Flow {
   constructor(msg: IMessage, connMgr: ConnectionManager) {
     this.no = ++Flow.curNo
     this.id = msg.id
+
+    this.addRequest(msg)
+
+    this._isTextRequest = null
+    this._isTextResponse = null
+    this._requestBody = null
+    this._responseBody = null
+
+    this.connMgr = connMgr
+  }
+
+  public addRequest(msg: IMessage): Flow {
+    this.status = MessageType.REQUEST
     this.waitIntercept = msg.waitIntercept
 
     const flowRequestMsg = msg.content as IFlowRequest
@@ -93,12 +106,7 @@ export class Flow {
     this.url = new URL(rawUrl)
     this.path = this.url.pathname + this.url.search
 
-    this._isTextRequest = null
-    this._isTextResponse = null
-    this._requestBody = null
-    this._responseBody = null
-
-    this.connMgr = connMgr
+    return this
   }
 
   public addRequestBody(msg: IMessage): Flow {

web/conn.go

@@ -66,8 +66,8 @@ func (c *concurrentConn) whenConnClose(connCtx *proxy.ConnContext) {
 	}
 }
 
-func (c *concurrentConn) writeMessage(msg *messageFlow, f *proxy.Flow) {
-	if c.isIntercpt(f, msg) {
+func (c *concurrentConn) writeMessageMayWait(msg *messageFlow, f *proxy.Flow) {
+	if c.isIntercpt(f, msg.mType) {
 		msg.waitIntercept = 1
 	}
 
@@ -80,7 +80,18 @@ func (c *concurrentConn) writeMessage(msg *messageFlow, f *proxy.Flow) {
 	}
 
 	if msg.waitIntercept == 1 {
-		c.waitIntercept(f, msg)
+		c.waitIntercept(f)
+	}
+}
+
+func (c *concurrentConn) writeMessage(msg *messageFlow) {
+	msg.waitIntercept = 0
+	c.mu.Lock()
+	err := c.conn.WriteMessage(websocket.BinaryMessage, msg.bytes())
+	c.mu.Unlock()
+	if err != nil {
+		log.Error(err)
+		return
 	}
 }
 
@@ -129,8 +140,8 @@ func (c *concurrentConn) initWaitChan(key string) chan interface{} {
 }
 
 // 是否拦截
-func (c *concurrentConn) isIntercpt(f *proxy.Flow, after *messageFlow) bool {
-	if after.mType != messageTypeRequestBody && after.mType != messageTypeResponseBody {
+func (c *concurrentConn) isIntercpt(f *proxy.Flow, mType messageType) bool {
+	if mType != messageTypeRequestBody && mType != messageTypeResponseBody {
 		return false
 	}
 
@@ -139,7 +150,7 @@ func (c *concurrentConn) isIntercpt(f *proxy.Flow, after *messageFlow) bool {
 	}
 
 	var action int
-	if after.mType == messageTypeRequestBody {
+	if mType == messageTypeRequestBody {
 		action = 1
 	} else {
 		action = 2
@@ -164,7 +175,7 @@ func (c *concurrentConn) isIntercpt(f *proxy.Flow, after *messageFlow) bool {
 }
 
 // 拦截
-func (c *concurrentConn) waitIntercept(f *proxy.Flow, after *messageFlow) {
+func (c *concurrentConn) waitIntercept(f *proxy.Flow) {
 	ch := c.initWaitChan(f.Id.String())
 	msg := (<-ch).(*messageEdit)
 

web/web.go

@@ -23,10 +23,14 @@ type WebAddon struct {
 
 	conns   []*concurrentConn
 	connsMu sync.RWMutex
+
+	flowMessageState map[*proxy.Flow]messageType
 }
 
 func NewWebAddon(addr string) *WebAddon {
-	web := &WebAddon{}
+	web := &WebAddon{
+		flowMessageState: make(map[*proxy.Flow]messageType),
+	}
 
 	_, err := assets.ReadDir("client/build/dist")
 	if err != nil {
@@ -117,7 +121,7 @@ func (web *WebAddon) forEachConn(do func(c *concurrentConn)) bool {
 	return true
 }
 
-func (web *WebAddon) sendFlow(f *proxy.Flow, msgFn func() *messageFlow) bool {
+func (web *WebAddon) sendFlowMayWait(f *proxy.Flow, msgFn func() *messageFlow) bool {
 	web.connsMu.RLock()
 	conns := web.conns
 	web.connsMu.RUnlock()
@@ -128,28 +132,36 @@ func (web *WebAddon) sendFlow(f *proxy.Flow, msgFn func() *messageFlow) bool {
 
 	msg := msgFn()
 	for _, c := range conns {
-		c.writeMessage(msg, f)
+		c.writeMessageMayWait(msg, f)
 	}
 
 	return true
 }
 
 func (web *WebAddon) Requestheaders(f *proxy.Flow) {
+	web.flowMessageState[f] = messageType(0)
+	go func() {
+		<-f.Done()
+		web.sendMessageUntil(f, messageTypeResponseBody)
+		delete(web.flowMessageState, f)
+	}()
+
 	if f.ConnContext.ClientConn.Tls {
 		web.forEachConn(func(c *concurrentConn) {
 			c.trySendConnMessage(f)
 		})
 	}
-
-	web.sendFlow(f, func() *messageFlow {
-		return newMessageFlow(messageTypeRequest, f)
-	})
 }
 
 func (web *WebAddon) Request(f *proxy.Flow) {
-	web.sendFlow(f, func() *messageFlow {
-		return newMessageFlow(messageTypeRequestBody, f)
-	})
+	if web.isIntercpt(f, messageTypeRequestBody) {
+		web.sendFlowMayWait(f, func() *messageFlow {
+			return newMessageFlow(messageTypeRequest, f)
+		})
+		web.sendFlowMayWait(f, func() *messageFlow {
+			return newMessageFlow(messageTypeRequestBody, f)
+		})
+	}
 }
 
 func (web *WebAddon) Responseheaders(f *proxy.Flow) {
@@ -159,19 +171,68 @@ func (web *WebAddon) Responseheaders(f *proxy.Flow) {
 		})
 	}
 
-	web.sendFlow(f, func() *messageFlow {
-		return newMessageFlow(messageTypeResponse, f)
-	})
+	web.sendMessageUntil(f, messageTypeRequestBody)
 }
 
 func (web *WebAddon) Response(f *proxy.Flow) {
-	web.sendFlow(f, func() *messageFlow {
-		return newMessageFlow(messageTypeResponseBody, f)
-	})
+	if web.isIntercpt(f, messageTypeResponseBody) {
+		web.sendFlowMayWait(f, func() *messageFlow {
+			return newMessageFlow(messageTypeResponse, f)
+		})
+		web.sendFlowMayWait(f, func() *messageFlow {
+			return newMessageFlow(messageTypeResponseBody, f)
+		})
+	}
 }
 
 func (web *WebAddon) ServerDisconnected(connCtx *proxy.ConnContext) {
 	web.forEachConn(func(c *concurrentConn) {
 		c.whenConnClose(connCtx)
 	})
 }
+
+func (web *WebAddon) isIntercpt(f *proxy.Flow, mType messageType) bool {
+	web.connsMu.RLock()
+	conns := web.conns
+	web.connsMu.RUnlock()
+
+	if len(conns) == 0 {
+		return false
+	}
+
+	for _, c := range conns {
+		if c.isIntercpt(f, mType) {
+			return true
+		}
+	}
+	return false
+}
+
+func (web *WebAddon) sendFlow(msgFn func() *messageFlow) bool {
+	web.connsMu.RLock()
+	conns := web.conns
+	web.connsMu.RUnlock()
+
+	if len(conns) == 0 {
+		return false
+	}
+
+	msg := msgFn()
+	for _, c := range conns {
+		c.writeMessage(msg)
+	}
+
+	return true
+}
+
+func (web *WebAddon) sendMessageUntil(f *proxy.Flow, mType messageType) {
+	if web.flowMessageState[f] >= mType {
+		return
+	}
+	for state := web.flowMessageState[f] + 1; state <= mType; state++ {
+		web.sendFlow(func() *messageFlow {
+			return newMessageFlow(state, f)
+		})
+	}
+	web.flowMessageState[f] = mType
+}