Merge pull request #142 from kcajmagic/consumerRetryLogic

added alternative urls and consumer max retry logic

CHANGELOG.md

@@ -5,6 +5,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added 
+- Add alternative urls and consumer max retry logic for webhooks [issue 140](https://github.com/Comcast/caduceus/pull/140)
+
 ### Changed
 - Retry on non 2xx status codes [issue 139](https://github.com/Comcast/caduceus/pull/139)
 - Fix for no retries being attempted [issue 141](https://github.com/Comcast/caduceus/pull/141)

src/caduceus/outboundSender.go

@@ -165,6 +165,11 @@ func (osf OutboundSenderFactory) New() (obs OutboundSender, err error) {
 		return
 	}
 
+	// update default deliver retry count for sender
+	if osf.Listener.Config.MaxRetryCount != 0 {
+		osf.DeliveryRetries = osf.Listener.Config.MaxRetryCount
+	}
+
 	caduceusOutboundSender := &CaduceusOutboundSender{
 		id:               osf.Listener.Config.URL,
 		listener:         osf.Listener,
@@ -263,6 +268,11 @@ func (obs *CaduceusOutboundSender) Update(wh webhook.W) (err error) {
 	obs.deliverUntil = wh.Until
 	obs.events = events
 
+	// update default deliver retry count for sender
+	if wh.Config.MaxRetryCount != 0 {
+		obs.deliveryRetries = wh.Config.MaxRetryCount
+	}
+
 	// if matcher list is empty set it nil for Queue() logic
 	obs.matcher = nil
 	if 0 < len(matcher) {
@@ -502,6 +512,23 @@ func (obs *CaduceusOutboundSender) send(secret, acceptType string, msg *wrp.Mess
 		},
 	}
 
+	// if the consumer request alternative urls update subsequent requests with the new urls.
+	if len(obs.listener.Config.AlternativeURLs) > 0 {
+		index := 0
+		retryOptions.UpdateRequest = func(request *http.Request) {
+			if index >= len(obs.listener.Config.AlternativeURLs) {
+				index = 0
+			}
+			url, err := url.Parse(obs.listener.Config.AlternativeURLs[index])
+			index++
+			if err != nil {
+				logging.Error(obs.logger).Log(logging.MessageKey(), "failed to update url", "url", obs.listener.Config.AlternativeURLs[index], logging.ErrorKey(), err)
+				return
+			}
+			request.URL = url
+		}
+	}
+
 	// Send it
 	resp, err := xhttp.RetryTransactor(retryOptions, obs.sender)(req)
 	code := "failure"

src/caduceus/outboundSender_test.go

@@ -273,6 +273,54 @@ func Test429Retry(t *testing.T) {
 	assert.Equal(int32(2), trans.i)
 }
 
+func TestAltURL(t *testing.T) {
+	assert := assert.New(t)
+
+	urls := map[string]int{}
+
+	w := webhook.W{
+		Until:  time.Now().Add(60 * time.Second),
+		Events: []string{".*"},
+	}
+	w.Config.URL = "http://localhost:9999/foo"
+	w.Config.ContentType = "application/json"
+	w.Config.MaxRetryCount = 3
+	w.Config.AlternativeURLs = []string{
+		"http://localhost:9999/bar",
+		"http://localhost:9999/faa",
+		"http://localhost:9999/bas",
+	}
+
+	trans := &transport{}
+	trans.fn = func(req *http.Request, count int) (*http.Response, error) {
+		if _, ok := urls[req.URL.Path]; ok {
+			urls[req.URL.Path]++
+		} else {
+			urls[req.URL.Path] = 1
+		}
+		return &http.Response{StatusCode: 429}, nil
+	}
+
+	obs, err := simpleSetup(trans, time.Second, nil)
+	assert.Nil(err)
+	err = obs.Update(w)
+	assert.NotNil(obs)
+	assert.Nil(err)
+
+	req := simpleRequest()
+	req.Source = "mac:112233445566"
+	req.TransactionUUID = "1234"
+	req.Destination = "event:iot"
+	obs.Queue(req)
+
+	obs.Shutdown(true)
+
+	assert.Equal(int32(4), trans.i)
+	for k, v := range urls {
+		assert.Equal(1, v, k)
+	}
+}
+
 // Simple test that covers the normal successful case with extra matchers
 func TestSimpleWrpWithMatchers(t *testing.T) {
 

src/glide.yaml

@@ -1,6 +1,6 @@
 package: .
 import:
 - package: github.com/Comcast/webpa-common
-  version: af2102421561cc50aa301642bb52a8376d834a61
+  version: v1.1.0
 - package: github.com/satori/go.uuid
   version: f58768cc1a7a7e77a3bd49e98cdd21419399b6a3