update filter

pkg/context/context.go

@@ -215,3 +215,8 @@ func (ctx *context) Finish() {
  fn()
  }
 }
+
+func (ctx *context) Span() tracing.Span {
+ // TODO: add span
+ return nil
+}

pkg/filters/apiaggregator/apiaggregator.go

@@ -33,8 +33,9 @@ import (
  "github.com/megaease/easegress/pkg/filters"
  "github.com/megaease/easegress/pkg/logger"
  "github.com/megaease/easegress/pkg/object/rawconfigtrafficcontroller"
+ "github.com/megaease/easegress/pkg/protocols/httpprot"
+ "github.com/megaease/easegress/pkg/protocols/httpprot/httpheader"
  "github.com/megaease/easegress/pkg/tracing"
- "github.com/megaease/easegress/pkg/util/httpheader"
  "github.com/megaease/easegress/pkg/util/pathadaptor"
 )
 
@@ -174,30 +175,32 @@ func (aa *APIAggregator) reload() {
 
 // Handle limits HTTPContext.
 func (aa *APIAggregator) Handle(ctx context.Context) (result string) {
+ httpreq := ctx.Request().(*httpprot.Request)
+ httpresp := ctx.Response().(*httpprot.Response)
  buff := bytes.NewBuffer(nil)
  if aa.spec.MaxBodyBytes > 0 {
- written, err := io.CopyN(buff, ctx.Request().Body(), aa.spec.MaxBodyBytes+1)
+ written, err := io.CopyN(buff, httpreq.Payload().NewReader(), aa.spec.MaxBodyBytes+1)
  if written > aa.spec.MaxBodyBytes {
  ctx.AddTag(fmt.Sprintf("apiAggregator: request body exceed %dB", aa.spec.MaxBodyBytes))
- ctx.Response().SetStatusCode(http.StatusRequestEntityTooLarge)
+ httpresp.SetStatusCode(http.StatusRequestEntityTooLarge)
  return resultFailed
  }
  if err != io.EOF {
  ctx.AddTag(fmt.Sprintf("apiAggregator: read request body failed: %v", err))
- ctx.Response().SetStatusCode(http.StatusBadRequest)
+ httpresp.SetStatusCode(http.StatusBadRequest)
  return resultFailed
  }
  }
 
  wg := &sync.WaitGroup{}
  wg.Add(len(aa.spec.Pipelines))
 
- httpResps := make([]context.HTTPResponse, len(aa.spec.Pipelines))
+ httpResps := make([]*httpprot.Response, len(aa.spec.Pipelines))
  for i, p := range aa.spec.Pipelines {
  req, err := aa.newHTTPReq(ctx, p, buff)
  if err != nil {
  logger.Errorf("BUG: new HTTP request failed: %v, pipelinename: %s", err, aa.spec.Pipelines[i].Name)
- ctx.Response().SetStatusCode(http.StatusBadRequest)
+ httpresp.SetStatusCode(http.StatusBadRequest)
  return resultFailed
  }
 
@@ -209,9 +212,9 @@ func (aa *APIAggregator) Handle(ctx context.Context) (result string) {
  return
  }
  w := httptest.NewRecorder()
- copyCtx := context.New(w, req, tracing.NoopTracing, "no trace")
+ copyCtx := context.New(httpprot.NewRequest(req), httpprot.NewResponse(w), tracing.NoopTracing, "no trace")
  handler.Handle(copyCtx)
- rsp := copyCtx.Response()
+ rsp := copyCtx.Response().(*httpprot.Response)
 
  if rsp != nil && rsp.StatusCode() == http.StatusOK {
  httpResps[i] = rsp
@@ -225,9 +228,7 @@ func (aa *APIAggregator) Handle(ctx context.Context) (result string) {
  _resp := resp
 
  if resp != nil {
- if body, ok := _resp.Body().(io.ReadCloser); ok {
- defer body.Close()
- }
+ _resp.Payload().Close()
  }
  }
 
@@ -236,14 +237,14 @@ func (aa *APIAggregator) Handle(ctx context.Context) (result string) {
  // Get all Pipeline response' body
  for i, resp := range httpResps {
  if resp == nil && !aa.spec.PartialSucceed {
- ctx.Response().Std().Header().Set("X-EG-Aggregator", fmt.Sprintf("failed-in-%s",
+ httpresp.Header().Set("X-EG-Aggregator", fmt.Sprintf("failed-in-%s",
  aa.spec.Pipelines[i].Name))
- ctx.Response().SetStatusCode(http.StatusServiceUnavailable)
+ httpresp.SetStatusCode(http.StatusServiceUnavailable)
  return resultFailed
  }
 
- if resp != nil && resp.Body() != nil {
- if res := aa.copyHTTPBody2Map(resp.Body(), ctx, data, aa.spec.Pipelines[i].Name); len(res) != 0 {
+ if resp != nil && resp.Payload().NewReader() != nil {
+ if res := aa.copyHTTPBody2Map(resp.Payload().NewReader(), ctx, data, aa.spec.Pipelines[i].Name); len(res) != 0 {
  return res
  }
  }
@@ -253,18 +254,20 @@ func (aa *APIAggregator) Handle(ctx context.Context) (result string) {
 }
 
 func (aa *APIAggregator) newHTTPReq(ctx context.Context, p *Pipeline, buff *bytes.Buffer) (*http.Request, error) {
- var stdctx stdcontext.Context = ctx
+ httpreq := ctx.Request().(*httpprot.Request)
+
+ var stdctx stdcontext.Context = httpreq.Context()
  if aa.spec.timeout != nil {
  // NOTE: Cancel function could be omitted here.
  stdctx, _ = stdcontext.WithTimeout(stdctx, *aa.spec.timeout)
  }
 
- method := ctx.Request().Method()
+ method := httpreq.Method()
  if p.Method != "" {
  method = p.Method
  }
 
- url := ctx.Request().Std().URL
+ url := httpreq.Std().URL
  if p.pa != nil {
  url.Path = p.pa.Adapt(url.Path)
  }
@@ -278,17 +281,18 @@ func (aa *APIAggregator) newHTTPReq(ctx context.Context, p *Pipeline, buff *byte
 }
 
 func (aa *APIAggregator) copyHTTPBody2Map(body io.Reader, ctx context.Context, data map[string][]byte, name string) string {
+ httpresp := ctx.Response().(*httpprot.Response)
  respBody := bytes.NewBuffer(nil)
 
  written, err := io.CopyN(respBody, body, aa.spec.MaxBodyBytes)
  if written > aa.spec.MaxBodyBytes {
  ctx.AddTag(fmt.Sprintf("apiAggregator: response body exceed %dB", aa.spec.MaxBodyBytes))
- ctx.Response().SetStatusCode(http.StatusInsufficientStorage)
+ httpresp.SetStatusCode(http.StatusInsufficientStorage)
  return resultFailed
  }
  if err != io.EOF {
  ctx.AddTag(fmt.Sprintf("apiAggregator: read response body failed: %v", err))
- ctx.Response().SetStatusCode(http.StatusInternalServerError)
+ httpresp.SetStatusCode(http.StatusInternalServerError)
  return resultFailed
  }
 
@@ -298,14 +302,15 @@ func (aa *APIAggregator) copyHTTPBody2Map(body io.Reader, ctx context.Context, d
 }
 
 func (aa *APIAggregator) formatResponse(ctx context.Context, data map[string][]byte) string {
+ httpresp := ctx.Response().(*httpprot.Response)
  if aa.spec.MergeResponse {
  result := map[string]interface{}{}
  for _, resp := range data {
  err := jsoniter.Unmarshal(resp, &result)
  if err != nil {
  ctx.AddTag(fmt.Sprintf("apiAggregator: unmarshal %s to json object failed: %v",
  resp, err))
- ctx.Response().SetStatusCode(context.EGStatusBadResponse)
+ httpresp.SetStatusCode(context.EGStatusBadResponse)
  return resultFailed
  }
  }
@@ -314,11 +319,11 @@ func (aa *APIAggregator) formatResponse(ctx context.Context, data map[string][]b
  ctx.AddTag(fmt.Sprintf("apiAggregator: marshal %#v to json failed: %v",
  result, err))
  logger.Errorf("apiAggregator: marshal %#v to json failed: %v", result, err)
- ctx.Response().SetStatusCode(http.StatusInternalServerError)
+ httpresp.SetStatusCode(http.StatusInternalServerError)
  return resultFailed
  }
 
- ctx.Response().SetBody(bytes.NewReader(buff))
+ httpresp.Payload().SetReader(bytes.NewReader(buff), true)
  } else {
  result := []map[string]interface{}{}
  for _, resp := range data {
@@ -327,7 +332,7 @@ func (aa *APIAggregator) formatResponse(ctx context.Context, data map[string][]b
  if err != nil {
  ctx.AddTag(fmt.Sprintf("apiAggregator: unmarshal %s to json object failed: %v",
  resp, err))
- ctx.Response().SetStatusCode(context.EGStatusBadResponse)
+ httpresp.SetStatusCode(context.EGStatusBadResponse)
  return resultFailed
  }
  result = append(result, ele)
@@ -336,11 +341,11 @@ func (aa *APIAggregator) formatResponse(ctx context.Context, data map[string][]b
  if err != nil {
  ctx.AddTag(fmt.Sprintf("apiAggregator: marshal %#v to json failed: %v",
  result, err))
- ctx.Response().SetStatusCode(http.StatusInternalServerError)
+ httpresp.SetStatusCode(http.StatusInternalServerError)
  return resultFailed
  }
 
- ctx.Response().SetBody(bytes.NewReader(buff))
+ httpresp.Payload().SetReader(bytes.NewReader(buff), true)
  }
 
  return ""

pkg/filters/bridge/bridge.go

@@ -23,6 +23,7 @@ import (
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/filters"
  "github.com/megaease/easegress/pkg/logger"
+ "github.com/megaease/easegress/pkg/protocols/httpprot"
 )
 
 const (
@@ -118,6 +119,7 @@ func (b *Bridge) InjectMuxMapper(mapper context.MuxMapper) {
 
 // Handle builds a bridge for pipeline.
 func (b *Bridge) Handle(ctx context.Context) (result string) {
+ httpresp := ctx.Response().(*httpprot.Response)
  if len(b.spec.Destinations) <= 0 {
  panic("not any destination defined")
  }
@@ -141,15 +143,15 @@ func (b *Bridge) Handle(ctx context.Context) (result string) {
 
  if !found {
  logger.Errorf("dest not found: %s", dest)
- ctx.Response().SetStatusCode(http.StatusServiceUnavailable)
+ httpresp.SetStatusCode(http.StatusServiceUnavailable)
  return resultDestinationNotFound
  }
 
  handler, exists := b.muxMapper.GetHandler(dest)
 
  if !exists {
  logger.Errorf("failed to get running object %s", b.spec.Destinations[0])
- ctx.Response().SetStatusCode(http.StatusServiceUnavailable)
+ httpresp.SetStatusCode(http.StatusServiceUnavailable)
  return resultDestinationNotFound
  }
 

pkg/filters/canary/rule.go

@@ -21,6 +21,7 @@ import (
  yaml "gopkg.in/yaml.v2"
 
  "github.com/megaease/easegress/pkg/context"
+ "github.com/megaease/easegress/pkg/protocols/httpprot"
 )
 
 // Rule is the A/B testing rule specified by user.
@@ -85,9 +86,10 @@ func (r *Rule) parse(p []byte) (err error) {
 // if true, add tag.
 // if false, do nothing.
 func (r *Rule) doMatch(ctx context.Context) {
+ httpreq := ctx.Request().(*httpprot.Request)
  if r.isMatch(&sourceData{
- req: ctx.Request().Std(),
- clientIP: ctx.Request().RealIP(),
+ req: httpreq.Std(),
+ clientIP: httpreq.RealIP(),
  }) {
  ctx.Request().Header().Set(r.TagKey, r.TagValue)
  }

pkg/filters/httprequestbuilder/httprequestbuilder.go

@@ -20,7 +20,6 @@ package requestbuilder
 import (
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/filters"
- "github.com/megaease/easegress/pkg/protocols"
 )
 
 const (
@@ -92,7 +91,8 @@ func (rb *HTTPRequestBuilder) reload() {
 
 // Handle builds request.
 func (rb *HTTPRequestBuilder) Handle(ctx context.Context) string {
- r := protocols.HTTP().CreateRequest().(*protocols.HTTPRequest)
+ // TODO: finish it later
+ panic("finish this when ready")
  return ""
 }
 

pkg/filters/httprequestbuilder/httprequestbuilder_test.go

@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package requestbuilder
+
+import "testing"
+
+func TestRequestBuilder(t *testing.T) {
+ // this test will panic 100% update it when ready
+ // don't forget TODO
+ rb := &HTTPRequestBuilder{}
+ rb.Handle(nil)
+}

pkg/filters/meshadaptor/meshadaptor.go

@@ -20,8 +20,9 @@ package meshadaptor
 import (
  "github.com/megaease/easegress/pkg/context"
  "github.com/megaease/easegress/pkg/filters"
- "github.com/megaease/easegress/pkg/util/httpfilter"
- "github.com/megaease/easegress/pkg/util/httpheader"
+ "github.com/megaease/easegress/pkg/protocols/httpprot"
+ "github.com/megaease/easegress/pkg/protocols/httpprot/httpfilter"
+ "github.com/megaease/easegress/pkg/protocols/httpprot/httpheader"
  "github.com/megaease/easegress/pkg/util/pathadaptor"
 )
 
@@ -104,14 +105,16 @@ func (ra *MeshAdaptor) reload() {
 
 // Handle adapts request.
 func (ra *MeshAdaptor) Handle(ctx context.Context) string {
+ httpreq := ctx.Request().(*httpprot.Request)
  for _, serviceCanary := range ra.spec.ServiceCanaries {
- if serviceCanary.filter.Filter(ctx) {
- ctx.Request().Header().Adapt(serviceCanary.Header, ctx.Template())
+ if serviceCanary.filter.Filter(httpreq) {
+ // ctx.Request().Header().Adapt(serviceCanary.Header, ctx.Template())
+ // TODO: add context template here!
+ panic("")
  }
  }
 
- result := ""
- return ctx.CallNextHandler(result)
+ return ""
 }
 
 // Status returns status.

pkg/filters/meshadaptor/meshadaptor_test.go

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2017, MegaEase
+ * All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http:https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package meshadaptor
+
+import "testing"
+
+func TestMeshAdaptor(t *testing.T) {
+ // this test will panic 100% only update this test when finish context template
+ ra := &MeshAdaptor{}
+ ra.Handle(nil)
+}

pkg/filters/mock/mock.go

@@ -198,6 +198,7 @@ func (m *Mock) match(ctx context.Context) *Rule {
 }
 
 func (m *Mock) mock(ctx context.Context, rule *Rule) {
+ httpreq := ctx.Request().(*httpprot.Request)
  w := ctx.Response().(*httpprot.Response)
  w.SetStatusCode(rule.Code)
  for key, value := range rule.Headers {
@@ -211,7 +212,7 @@ func (m *Mock) mock(ctx context.Context, rule *Rule) {
 
  logger.Debugf("delay for %v ...", rule.delay)
  select {
- case <-ctx.Done():
+ case <-httpreq.Context().Done():
  logger.Debugf("request cancelled in the middle of delay mocking")
  case <-time.After(rule.delay):
  }