Event persistence for MQTT (#7268 )

- The events will be persisted in queueStore if `queueDir` is set. - Else, if queueDir is not set events persist in memory. The events are replayed back when the mqtt broker is back online.
fix json syntax (#7285 )
2026-11-07 16:22:05 +03:00 · 2019-02-25 18:01:13 -08:00 · 2019-02-25 13:02:59 +05:30 · 2019-02-24 18:50:28 -08:00 · 2019-02-24 06:51:02 -08:00 · 2019-02-24 11:44:24 +05:30
44 changed files with 1147 additions and 335 deletions
@@ -66,7 +66,7 @@ coverage: build
 build: checks
 	@echo "Building minio binary to './minio'"
 	@GOFLAGS="" CGO_ENABLED=0 go build -tags kqueue --ldflags $(BUILD_LDFLAGS) -o $(PWD)/minio
-	@GOFLAGS="" CGO_ENABLED=0 go build --ldflags="-s -w" -o $(PWD)/dockerscripts/healthcheck $(PWD)/dockerscripts/healthcheck.go
+	@GOFLAGS="" CGO_ENABLED=0 go build -tags kqueue --ldflags="-s -w" -o $(PWD)/dockerscripts/healthcheck $(PWD)/dockerscripts/healthcheck.go

 docker: build
 	@docker build -t $(TAG) . -f Dockerfile.dev
@@ -122,7 +122,8 @@ var (
        "password": "",
        "reconnectInterval": 0,
 	"keepAliveInterval": 0,
-	"queueDir": ""
+	"queueDir": "",
+        "queueLimit": 0
      }
    },
    "mysql": {
@@ -73,6 +73,7 @@ const (
 	ErrInvalidCopyPartRange
 	ErrInvalidCopyPartRangeSource
 	ErrInvalidMaxKeys
+	ErrInvalidEncodingMethod
 	ErrInvalidMaxUploads
 	ErrInvalidMaxParts
 	ErrInvalidPartNumberMarker
@@ -357,6 +358,11 @@ var errorCodes = errorCodeMap{
 		Description:    "Argument maxKeys must be an integer between 0 and 2147483647",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrInvalidEncodingMethod: {
+		Code:           "InvalidArgument",
+		Description:    "Invalid Encoding Method specified in Request",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrInvalidMaxParts: {
 		Code:           "InvalidArgument",
 		Description:    "Argument max-parts must be an integer between 0 and 2147483647",
@@ -1,60 +0,0 @@
-/*
- * Minio Cloud Storage, (C) 2016 Minio, Inc.
- *
- * 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://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 cmd
-
-import (
-	"context"
-	"net/http"
-)
-
-// Represents additional fields necessary for ErrPartTooSmall S3 error.
-type completeMultipartAPIError struct {
-	// Proposed size represents uploaded size of the part.
-	ProposedSize int64
-	// Minimum size allowed epresents the minimum size allowed per
-	// part. Defaults to 5MB.
-	MinSizeAllowed int64
-	// Part number of the part which is incorrect.
-	PartNumber int
-	// ETag of the part which is incorrect.
-	PartETag string
-	// Other default XML error responses.
-	APIErrorResponse
-}
-
-// writeErrorResponsePartTooSmall - function is used specifically to
-// construct a proper error response during CompleteMultipartUpload
-// when one of the parts is < 5MB.
-// The requirement comes due to the fact that generic ErrorResponse
-// XML doesn't carry the additional fields required to send this
-// error. So we construct a new type which lies well within the scope
-// of this function.
-func writePartSmallErrorResponse(ctx context.Context, w http.ResponseWriter, r *http.Request, err PartTooSmall) {
-	apiError := toAPIError(ctx, err)
-	// Generate complete multipart error response.
-	errorResponse := getAPIErrorResponse(ctx, apiError, r.URL.Path,
-		w.Header().Get(responseRequestIDKey),
-		w.Header().Get(responseDeploymentIDKey))
-	cmpErrResp := completeMultipartAPIError{err.PartSize, int64(5242880), err.PartNumber, err.PartETag, errorResponse}
-	encodedErrorResponse := encodeResponse(cmpErrResp)
-
-	// respond with 400 bad request.
-	w.WriteHeader(apiError.HTTPStatusCode)
-	// Write error body.
-	w.Write(encodedErrorResponse)
-	w.(http.Flusher).Flush()
-}
@@ -280,7 +280,7 @@ func getURLScheme(tls bool) string {
 }

 // getObjectLocation gets the fully qualified URL of an object.
-func getObjectLocation(r *http.Request, domain, bucket, object string) string {
+func getObjectLocation(r *http.Request, domains []string, bucket, object string) string {
 	// unit tests do not have host set.
 	if r.Host == "" {
 		return path.Clean(r.URL.Path)
@@ -295,15 +295,31 @@ func getObjectLocation(r *http.Request, domain, bucket, object string) string {
 		Scheme: proto,
 	}
 	// If domain is set then we need to use bucket DNS style.
-	if domain != "" {
+	for _, domain := range domains {
 		if strings.Contains(r.Host, domain) {
 			u.Host = bucket + "." + r.Host
 			u.Path = path.Join(slashSeparator, object)
+			break
 		}
 	}
 	return u.String()
 }

+// s3EncodeName encodes string in response when encodingType
+// is specified in AWS S3 requests.
+func s3EncodeName(name string, encodingType string) (result string) {
+	// Quick path to exit
+	if encodingType == "" {
+		return name
+	}
+	encodingType = strings.ToLower(encodingType)
+	switch encodingType {
+	case "url":
+		return url.QueryEscape(name)
+	}
+	return name
+}
+
 // generates ListBucketsResponse from array of BucketInfo which can be
 // serialized to match XML and JSON API spec output.
 func generateListBucketsResponse(buckets []BucketInfo) ListBucketsResponse {
@@ -326,7 +342,7 @@ func generateListBucketsResponse(buckets []BucketInfo) ListBucketsResponse {
 }

 // generates an ListObjectsV1 response for the said bucket with other enumerated options.
-func generateListObjectsV1Response(bucket, prefix, marker, delimiter string, maxKeys int, resp ListObjectsInfo) ListObjectsResponse {
+func generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType string, maxKeys int, resp ListObjectsInfo) ListObjectsResponse {
 	var contents []Object
 	var prefixes []CommonPrefix
 	var owner = Owner{}
@@ -338,7 +354,7 @@ func generateListObjectsV1Response(bucket, prefix, marker, delimiter string, max
 		if object.Name == "" {
 			continue
 		}
-		content.Key = object.Name
+		content.Key = s3EncodeName(object.Name, encodingType)
 		content.LastModified = object.ModTime.UTC().Format(timeFormatAMZLong)
 		if object.ETag != "" {
 			content.ETag = "\"" + object.ETag + "\""
@@ -348,20 +364,20 @@ func generateListObjectsV1Response(bucket, prefix, marker, delimiter string, max
 		content.Owner = owner
 		contents = append(contents, content)
 	}
-	// TODO - support EncodingType in xml decoding
 	data.Name = bucket
 	data.Contents = contents

-	data.Prefix = prefix
-	data.Marker = marker
-	data.Delimiter = delimiter
+	data.EncodingType = encodingType
+	data.Prefix = s3EncodeName(prefix, encodingType)
+	data.Marker = s3EncodeName(marker, encodingType)
+	data.Delimiter = s3EncodeName(delimiter, encodingType)
 	data.MaxKeys = maxKeys

-	data.NextMarker = resp.NextMarker
+	data.NextMarker = s3EncodeName(resp.NextMarker, encodingType)
 	data.IsTruncated = resp.IsTruncated
 	for _, prefix := range resp.Prefixes {
 		var prefixItem = CommonPrefix{}
-		prefixItem.Prefix = prefix
+		prefixItem.Prefix = s3EncodeName(prefix, encodingType)
 		prefixes = append(prefixes, prefixItem)
 	}
 	data.CommonPrefixes = prefixes
@@ -369,7 +385,7 @@ func generateListObjectsV1Response(bucket, prefix, marker, delimiter string, max
 }

 // generates an ListObjectsV2 response for the said bucket with other enumerated options.
-func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter, delimiter string, fetchOwner, isTruncated bool, maxKeys int, objects []ObjectInfo, prefixes []string) ListObjectsV2Response {
+func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter, delimiter, encodingType string, fetchOwner, isTruncated bool, maxKeys int, objects []ObjectInfo, prefixes []string) ListObjectsV2Response {
 	var contents []Object
 	var commonPrefixes []CommonPrefix
 	var owner = Owner{}
@@ -384,7 +400,7 @@ func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter,
 		if object.Name == "" {
 			continue
 		}
-		content.Key = object.Name
+		content.Key = s3EncodeName(object.Name, encodingType)
 		content.LastModified = object.ModTime.UTC().Format(timeFormatAMZLong)
 		if object.ETag != "" {
 			content.ETag = "\"" + object.ETag + "\""
@@ -394,20 +410,20 @@ func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter,
 		content.Owner = owner
 		contents = append(contents, content)
 	}
-	// TODO - support EncodingType in xml decoding
 	data.Name = bucket
 	data.Contents = contents

-	data.StartAfter = startAfter
-	data.Delimiter = delimiter
-	data.Prefix = prefix
+	data.EncodingType = encodingType
+	data.StartAfter = s3EncodeName(startAfter, encodingType)
+	data.Delimiter = s3EncodeName(delimiter, encodingType)
+	data.Prefix = s3EncodeName(prefix, encodingType)
 	data.MaxKeys = maxKeys
 	data.ContinuationToken = token
 	data.NextContinuationToken = nextToken
 	data.IsTruncated = isTruncated
 	for _, prefix := range prefixes {
 		var prefixItem = CommonPrefix{}
-		prefixItem.Prefix = prefix
+		prefixItem.Prefix = s3EncodeName(prefix, encodingType)
 		commonPrefixes = append(commonPrefixes, prefixItem)
 	}
 	data.CommonPrefixes = commonPrefixes
@@ -451,11 +467,10 @@ func generateCompleteMultpartUploadResponse(bucket, key, location, etag string)
 }

 // generates ListPartsResponse from ListPartsInfo.
-func generateListPartsResponse(partsInfo ListPartsInfo) ListPartsResponse {
-	// TODO - support EncodingType in xml decoding
+func generateListPartsResponse(partsInfo ListPartsInfo, encodingType string) ListPartsResponse {
 	listPartsResponse := ListPartsResponse{}
 	listPartsResponse.Bucket = partsInfo.Bucket
-	listPartsResponse.Key = partsInfo.Object
+	listPartsResponse.Key = s3EncodeName(partsInfo.Object, encodingType)
 	listPartsResponse.UploadID = partsInfo.UploadID
 	listPartsResponse.StorageClass = globalMinioDefaultStorageClass
 	listPartsResponse.Initiator.ID = globalMinioDefaultOwnerID
@@ -479,29 +494,29 @@ func generateListPartsResponse(partsInfo ListPartsInfo) ListPartsResponse {
 }

 // generates ListMultipartUploadsResponse for given bucket and ListMultipartsInfo.
-func generateListMultipartUploadsResponse(bucket string, multipartsInfo ListMultipartsInfo) ListMultipartUploadsResponse {
+func generateListMultipartUploadsResponse(bucket string, multipartsInfo ListMultipartsInfo, encodingType string) ListMultipartUploadsResponse {
 	listMultipartUploadsResponse := ListMultipartUploadsResponse{}
 	listMultipartUploadsResponse.Bucket = bucket
-	listMultipartUploadsResponse.Delimiter = multipartsInfo.Delimiter
+	listMultipartUploadsResponse.Delimiter = s3EncodeName(multipartsInfo.Delimiter, encodingType)
 	listMultipartUploadsResponse.IsTruncated = multipartsInfo.IsTruncated
-	listMultipartUploadsResponse.EncodingType = multipartsInfo.EncodingType
-	listMultipartUploadsResponse.Prefix = multipartsInfo.Prefix
-	listMultipartUploadsResponse.KeyMarker = multipartsInfo.KeyMarker
-	listMultipartUploadsResponse.NextKeyMarker = multipartsInfo.NextKeyMarker
+	listMultipartUploadsResponse.EncodingType = encodingType
+	listMultipartUploadsResponse.Prefix = s3EncodeName(multipartsInfo.Prefix, encodingType)
+	listMultipartUploadsResponse.KeyMarker = s3EncodeName(multipartsInfo.KeyMarker, encodingType)
+	listMultipartUploadsResponse.NextKeyMarker = s3EncodeName(multipartsInfo.NextKeyMarker, encodingType)
 	listMultipartUploadsResponse.MaxUploads = multipartsInfo.MaxUploads
 	listMultipartUploadsResponse.NextUploadIDMarker = multipartsInfo.NextUploadIDMarker
 	listMultipartUploadsResponse.UploadIDMarker = multipartsInfo.UploadIDMarker
 	listMultipartUploadsResponse.CommonPrefixes = make([]CommonPrefix, len(multipartsInfo.CommonPrefixes))
 	for index, commonPrefix := range multipartsInfo.CommonPrefixes {
 		listMultipartUploadsResponse.CommonPrefixes[index] = CommonPrefix{
-			Prefix: commonPrefix,
+			Prefix: s3EncodeName(commonPrefix, encodingType),
 		}
 	}
 	listMultipartUploadsResponse.Uploads = make([]Upload, len(multipartsInfo.Uploads))
 	for index, upload := range multipartsInfo.Uploads {
 		newUpload := Upload{}
 		newUpload.UploadID = upload.UploadID
-		newUpload.Key = upload.Object
+		newUpload.Key = s3EncodeName(upload.Object, encodingType)
 		newUpload.Initiated = upload.Initiated.UTC().Format(timeFormatAMZLong)
 		listMultipartUploadsResponse.Uploads[index] = newUpload
 	}
@@ -587,7 +602,8 @@ func writeErrorResponse(ctx context.Context, w http.ResponseWriter, err APIError
 	}

 	// Generate error response.
-	errorResponse := getAPIErrorResponse(ctx, err, reqURL.Path, w.Header().Get(responseRequestIDKey), w.Header().Get(responseDeploymentIDKey))
+	errorResponse := getAPIErrorResponse(ctx, err, reqURL.Path,
+		w.Header().Get(responseRequestIDKey), w.Header().Get(responseDeploymentIDKey))
 	encodedErrorResponse := encodeResponse(errorResponse)
 	writeResponse(w, err.HTTPStatusCode, encodedErrorResponse, mimeXML)
 }
@@ -26,7 +26,7 @@ func TestObjectLocation(t *testing.T) {
 	testCases := []struct {
 		request          *http.Request
 		bucket, object   string
-		domain           string
+		domains          []string
 		expectedLocation string
 	}{
 		// Server binding to localhost IP with https.
@@ -80,7 +80,7 @@ func TestObjectLocation(t *testing.T) {
 				Host:   "mys3.bucket.org",
 				Header: map[string][]string{},
 			},
-			domain:           "mys3.bucket.org",
+			domains:          []string{"mys3.bucket.org"},
 			bucket:           "mybucket",
 			object:           "test/1.txt",
 			expectedLocation: "http://mybucket.mys3.bucket.org/test/1.txt",
@@ -92,14 +92,14 @@ func TestObjectLocation(t *testing.T) {
 					"X-Forwarded-Scheme": {httpsScheme},
 				},
 			},
-			domain:           "mys3.bucket.org",
+			domains:          []string{"mys3.bucket.org"},
 			bucket:           "mybucket",
 			object:           "test/1.txt",
 			expectedLocation: "https://mybucket.mys3.bucket.org/test/1.txt",
 		},
 	}
 	for i, testCase := range testCases {
-		gotLocation := getObjectLocation(testCase.request, testCase.domain, testCase.bucket, testCase.object)
+		gotLocation := getObjectLocation(testCase.request, testCase.domains, testCase.bucket, testCase.object)
 		if testCase.expectedLocation != gotLocation {
 			t.Errorf("Test %d: expected %s, got %s", i+1, testCase.expectedLocation, gotLocation)
 		}
@@ -44,8 +44,8 @@ func registerAPIRouter(router *mux.Router, encryptionEnabled bool) {
 	// API Router
 	apiRouter := router.PathPrefix("/").Subrouter()
 	var routers []*mux.Router
-	if globalDomainName != "" {
-		routers = append(routers, apiRouter.Host("{bucket:.+}."+globalDomainName).Subrouter())
+	for _, domainName := range globalDomainNames {
+		routers = append(routers, apiRouter.Host("{bucket:.+}."+domainName).Subrouter())
 	}
 	routers = append(routers, apiRouter.PathPrefix("/{bucket}").Subrouter())

@@ -18,6 +18,7 @@ package cmd

 import (
 	"net/http"
+	"strings"

 	"github.com/gorilla/mux"
 	"github.com/minio/minio/cmd/crypto"
@@ -32,12 +33,19 @@ import (
 // - delimiter if set should be equal to '/', otherwise the request is rejected.
 // - marker if set should have a common prefix with 'prefix' param, otherwise
 //   the request is rejected.
-func validateListObjectsArgs(prefix, marker, delimiter string, maxKeys int) APIErrorCode {
+func validateListObjectsArgs(prefix, marker, delimiter, encodingType string, maxKeys int) APIErrorCode {
 	// Max keys cannot be negative.
 	if maxKeys < 0 {
 		return ErrInvalidMaxKeys
 	}

+	if encodingType != "" {
+		// Only url encoding type is supported
+		if strings.ToLower(encodingType) != "url" {
+			return ErrInvalidEncodingMethod
+		}
+	}
+
 	/// Minio special conditions for ListObjects.

 	// Verify if delimiter is anything other than '/', which we do not support.
@@ -78,7 +86,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 	urlValues := r.URL.Query()

 	// Extract all the listObjectsV2 query params to their native values.
-	prefix, token, startAfter, delimiter, fetchOwner, maxKeys, _, errCode := getListObjectsV2Args(urlValues)
+	prefix, token, startAfter, delimiter, fetchOwner, maxKeys, encodingType, errCode := getListObjectsV2Args(urlValues)
 	if errCode != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(errCode), r.URL, guessIsBrowserReq(r))
 		return
@@ -86,7 +94,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http

 	// Validate the query params before beginning to serve the request.
 	// fetch-owner is not validated since it is a boolean
-	if s3Error := validateListObjectsArgs(prefix, token, delimiter, maxKeys); s3Error != ErrNone {
+	if s3Error := validateListObjectsArgs(prefix, token, delimiter, encodingType, maxKeys); s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
 		return
 	}
@@ -126,7 +134,7 @@ func (api objectAPIHandlers) ListObjectsV2Handler(w http.ResponseWriter, r *http
 	}

 	response := generateListObjectsV2Response(bucket, prefix, token, listObjectsV2Info.NextContinuationToken, startAfter,
-		delimiter, fetchOwner, listObjectsV2Info.IsTruncated, maxKeys, listObjectsV2Info.Objects, listObjectsV2Info.Prefixes)
+		delimiter, encodingType, fetchOwner, listObjectsV2Info.IsTruncated, maxKeys, listObjectsV2Info.Objects, listObjectsV2Info.Prefixes)

 	// Write success response.
 	writeSuccessResponseXML(w, encodeResponse(response))
@@ -158,14 +166,14 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 	}

 	// Extract all the litsObjectsV1 query params to their native values.
-	prefix, marker, delimiter, maxKeys, _, s3Error := getListObjectsV1Args(r.URL.Query())
+	prefix, marker, delimiter, maxKeys, encodingType, s3Error := getListObjectsV1Args(r.URL.Query())
 	if s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
 		return
 	}

 	// Validate all the query params before beginning to serve the request.
-	if s3Error := validateListObjectsArgs(prefix, marker, delimiter, maxKeys); s3Error != ErrNone {
+	if s3Error := validateListObjectsArgs(prefix, marker, delimiter, encodingType, maxKeys); s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
 		return
 	}
@@ -204,7 +212,7 @@ func (api objectAPIHandlers) ListObjectsV1Handler(w http.ResponseWriter, r *http
 			}
 		}
 	}
-	response := generateListObjectsV1Response(bucket, prefix, marker, delimiter, maxKeys, listObjectsInfo)
+	response := generateListObjectsV1Response(bucket, prefix, marker, delimiter, encodingType, maxKeys, listObjectsInfo)

 	// Write success response.
 	writeSuccessResponseXML(w, encodeResponse(response))
@@ -155,7 +155,7 @@ func (api objectAPIHandlers) ListMultipartUploadsHandler(w http.ResponseWriter,
 		return
 	}

-	prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, _, errCode := getBucketMultipartResources(r.URL.Query())
+	prefix, keyMarker, uploadIDMarker, delimiter, maxUploads, encodingType, errCode := getBucketMultipartResources(r.URL.Query())
 	if errCode != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(errCode), r.URL, guessIsBrowserReq(r))
 		return
@@ -180,7 +180,7 @@ func (api objectAPIHandlers) ListMultipartUploadsHandler(w http.ResponseWriter,
 		return
 	}
 	// generate response
-	response := generateListMultipartUploadsResponse(bucket, listMultipartsInfo)
+	response := generateListMultipartUploadsResponse(bucket, listMultipartsInfo, encodingType)
 	encodedSuccessResponse := encodeResponse(response)

 	// write success response.
@@ -446,7 +446,7 @@ func (api objectAPIHandlers) PutBucketHandler(w http.ResponseWriter, r *http.Req
 				}

 				// Make sure to add Location information here only for bucket
-				w.Header().Set("Location", getObjectLocation(r, globalDomainName, bucket, ""))
+				w.Header().Set("Location", getObjectLocation(r, globalDomainNames, bucket, ""))

 				writeSuccessResponseHeadersOnly(w)
 				return
@@ -505,7 +505,7 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentLength), r.URL, guessIsBrowserReq(r))
 		return
 	}
-	resource, err := getResource(r.URL.Path, r.Host, globalDomainName)
+	resource, err := getResource(r.URL.Path, r.Host, globalDomainNames)
 	if err != nil {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL, guessIsBrowserReq(r))
 		return
@@ -677,7 +677,7 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
 		return
 	}

-	location := getObjectLocation(r, globalDomainName, bucket, object)
+	location := getObjectLocation(r, globalDomainNames, bucket, object)
 	w.Header().Set("ETag", `"`+objInfo.ETag+`"`)
 	w.Header().Set("Location", location)

@@ -262,10 +262,14 @@ func handleCommonEnvVars() {
 		logger.FatalIf(err, "Unable to initialize etcd with %s", etcdEndpoints)
 	}

-	globalDomainName, globalIsEnvDomainName = os.LookupEnv("MINIO_DOMAIN")
-	if globalDomainName != "" {
-		if _, ok = dns2.IsDomainName(globalDomainName); !ok {
-			logger.Fatal(uiErrInvalidDomainValue(nil).Msg("Unknown value `%s`", globalDomainName), "Invalid MINIO_DOMAIN value in environment variable")
+	v, ok := os.LookupEnv("MINIO_DOMAIN")
+	if ok {
+		for _, domainName := range strings.Split(v, ",") {
+			if _, ok = dns2.IsDomainName(domainName); !ok {
+				logger.Fatal(uiErrInvalidDomainValue(nil).Msg("Unknown value `%s`", domainName),
+					"Invalid MINIO_DOMAIN value in environment variable")
+			}
+			globalDomainNames = append(globalDomainNames, domainName)
 		}
 	}

@@ -294,10 +298,10 @@ func handleCommonEnvVars() {
 		updateDomainIPs(localIP4)
 	}

-	if globalDomainName != "" && !globalDomainIPs.IsEmpty() && globalEtcdClient != nil {
+	if len(globalDomainNames) != 0 && !globalDomainIPs.IsEmpty() && globalEtcdClient != nil {
 		var err error
-		globalDNSConfig, err = dns.NewCoreDNS(globalDomainName, globalDomainIPs, globalMinioPort, globalEtcdClient)
-		logger.FatalIf(err, "Unable to initialize DNS config for %s.", globalDomainName)
+		globalDNSConfig, err = dns.NewCoreDNS(globalDomainNames, globalDomainIPs, globalMinioPort, globalEtcdClient)
+		logger.FatalIf(err, "Unable to initialize DNS config for %s.", globalDomainNames)
 	}

 	if drives := os.Getenv("MINIO_CACHE_DRIVES"); drives != "" {
@@ -130,8 +130,8 @@ func TestServerConfigWithEnvs(t *testing.T) {
 	}

 	// Check if serverConfig has the correct domain
-	if globalDomainName != "domain.com" {
-		t.Errorf("Expected Domain to be `domain.com`, found " + globalDomainName)
+	if globalDomainNames[0] != "domain.com" {
+		t.Errorf("Expected Domain to be `domain.com`, found " + globalDomainNames[0])
 	}
 }

@@ -233,7 +233,7 @@ func TestValidateConfig(t *testing.T) {
 		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "redis": { "1": { "enable": true, "format": "namespace", "address": "example.com:80", "password": "xxx", "key": "key1" } }}}`, true},

 		// Test 27 - Test MQTT
-		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "mqtt": { "1": { "enable": true, "broker": "",  "topic": "", "qos": 0, "username": "", "password": "", "queueDir": ""}}}}`, false},
+		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "mqtt": { "1": { "enable": true, "broker": "",  "topic": "", "qos": 0, "username": "", "password": "", "queueDir": "", "queueLimit": 0}}}}`, false},

 		// Test 28 - Test NSQ
 		{`{"version": "` + v + `", "credential": { "accessKey": "minio", "secretKey": "minio123" }, "region": "us-east-1", "browser": "on", "notify": { "nsq": { "1": { "enable": true, "nsqdAddress": "", "topic": ""} }}}`, false},
@@ -884,7 +884,7 @@ type serverConfigV32 struct {
 	} `json:"policy"`
 }

-// serverConfigV33 is just like version '32', removes clientID from NATS and MQTT, and adds queueDir with MQTT.
+// serverConfigV33 is just like version '32', removes clientID from NATS and MQTT, and adds queueDir, queueLimit with MQTT.
 type serverConfigV33 struct {
 	quick.Config `json:"-"` // ignore interfaces

@@ -205,7 +205,6 @@ func (endpoints EndpointList) GetString(i int) string {
 func localEndpointsMemUsage(endpoints EndpointList) ServerMemUsageInfo {
 	var memUsages []mem.Usage
 	var historicUsages []mem.Usage
-	var addr string
 	scratchSpace := map[string]bool{}
 	for _, endpoint := range endpoints {
 		// Only proceed for local endpoints
@@ -213,16 +212,13 @@ func localEndpointsMemUsage(endpoints EndpointList) ServerMemUsageInfo {
 			if _, ok := scratchSpace[endpoint.Host]; ok {
 				continue
 			}
-			addr = GetLocalPeer(endpoints)
-			memUsage := mem.GetUsage()
-			memUsages = append(memUsages, memUsage)
-			historicUsage := mem.GetHistoricUsage()
-			historicUsages = append(historicUsages, historicUsage)
+			memUsages = append(memUsages, mem.GetUsage())
+			historicUsages = append(historicUsages, mem.GetHistoricUsage())
 			scratchSpace[endpoint.Host] = true
 		}
 	}
 	return ServerMemUsageInfo{
-		Addr:          addr,
+		Addr:          GetLocalPeer(endpoints),
 		Usage:         memUsages,
 		HistoricUsage: historicUsages,
 	}
@@ -233,7 +229,6 @@ func localEndpointsMemUsage(endpoints EndpointList) ServerMemUsageInfo {
 func localEndpointsCPULoad(endpoints EndpointList) ServerCPULoadInfo {
 	var cpuLoads []cpu.Load
 	var historicLoads []cpu.Load
-	var addr string
 	scratchSpace := map[string]bool{}
 	for _, endpoint := range endpoints {
 		// Only proceed for local endpoints
@@ -241,16 +236,13 @@ func localEndpointsCPULoad(endpoints EndpointList) ServerCPULoadInfo {
 			if _, ok := scratchSpace[endpoint.Host]; ok {
 				continue
 			}
-			addr = GetLocalPeer(endpoints)
-			cpuLoad := cpu.GetLoad()
-			cpuLoads = append(cpuLoads, cpuLoad)
-			historicLoad := cpu.GetHistoricLoad()
-			historicLoads = append(historicLoads, historicLoad)
+			cpuLoads = append(cpuLoads, cpu.GetLoad())
+			historicLoads = append(historicLoads, cpu.GetHistoricLoad())
 			scratchSpace[endpoint.Host] = true
 		}
 	}
 	return ServerCPULoadInfo{
-		Addr:         addr,
+		Addr:         GetLocalPeer(endpoints),
 		Load:         cpuLoads,
 		HistoricLoad: historicLoads,
 	}
@@ -260,26 +252,22 @@ func localEndpointsCPULoad(endpoints EndpointList) ServerCPULoadInfo {
 // local endpoints from given list of endpoints
 func localEndpointsDrivePerf(endpoints EndpointList) ServerDrivesPerfInfo {
 	var dps []disk.Performance
-	var addr string
 	for _, endpoint := range endpoints {
 		// Only proceed for local endpoints
 		if endpoint.IsLocal {
-			addr = GetLocalPeer(endpoints)
 			if _, err := os.Stat(endpoint.Path); err != nil {
 				// Since this drive is not available, add relevant details and proceed
 				dps = append(dps, disk.Performance{Path: endpoint.Path, Error: err.Error()})
 				continue
 			}
-			tempObj := mustGetUUID()
-			fsPath := pathJoin(endpoint.Path, minioMetaTmpBucket, tempObj)
-			dp := disk.GetPerformance(fsPath)
+			dp := disk.GetPerformance(pathJoin(endpoint.Path, minioMetaTmpBucket, mustGetUUID()))
 			dp.Path = endpoint.Path
 			dps = append(dps, dp)
 		}
 	}

 	return ServerDrivesPerfInfo{
-		Addr: addr,
+		Addr: GetLocalPeer(endpoints),
 		Perf: dps,
 	}
 }
@@ -623,7 +623,7 @@ type bucketForwardingHandler struct {
 }

 func (f bucketForwardingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	if globalDNSConfig == nil || globalDomainName == "" ||
+	if globalDNSConfig == nil || len(globalDomainNames) == 0 ||
 		guessIsHealthCheckReq(r) || guessIsMetricsReq(r) ||
 		guessIsRPCReq(r) || isAdminReq(r) {
 		f.handler.ServeHTTP(w, r)
@@ -632,7 +632,7 @@ func (f bucketForwardingHandler) ServeHTTP(w http.ResponseWriter, r *http.Reques

 	// For browser requests, when federation is setup we need to
 	// specifically handle download and upload for browser requests.
-	if guessIsBrowserReq(r) && globalDNSConfig != nil && globalDomainName != "" {
+	if guessIsBrowserReq(r) && globalDNSConfig != nil && len(globalDomainNames) > 0 {
 		var bucket, _ string
 		switch r.Method {
 		case http.MethodPut:
@@ -33,7 +33,7 @@ import (
 	"github.com/minio/minio/pkg/auth"
 	"github.com/minio/minio/pkg/certs"
 	"github.com/minio/minio/pkg/dns"
-	"github.com/minio/minio/pkg/iam/policy"
+	iampolicy "github.com/minio/minio/pkg/iam/policy"
 	"github.com/minio/minio/pkg/iam/validator"
 )

@@ -175,9 +175,8 @@ var (
 	globalActiveCred  auth.Credentials
 	globalPublicCerts []*x509.Certificate

-	globalIsEnvDomainName bool
-	globalDomainName      string        // Root domain for virtual host style requests
-	globalDomainIPs       set.StringSet // Root domain IP address(s) for a distributed Minio deployment
+	globalDomainNames []string      // Root domains for virtual host style requests
+	globalDomainIPs   set.StringSet // Root domain IP address(s) for a distributed Minio deployment

 	globalListingTimeout   = newDynamicTimeout( /*30*/ 600*time.Second /*5*/, 600*time.Second) // timeout for listing related ops
 	globalObjectTimeout    = newDynamicTimeout( /*1*/ 10*time.Minute /*10*/, 600*time.Second)  // timeout for Object API related ops
@@ -341,8 +341,8 @@ func httpTraceHdrs(f http.HandlerFunc) http.HandlerFunc {
 }

 // Returns "/bucketName/objectName" for path-style or virtual-host-style requests.
-func getResource(path string, host string, domain string) (string, error) {
-	if domain == "" {
+func getResource(path string, host string, domains []string) (string, error) {
+	if len(domains) == 0 {
 		return path, nil
 	}
 	// If virtual-host-style is enabled construct the "resource" properly.
@@ -357,11 +357,14 @@ func getResource(path string, host string, domain string) (string, error) {
 			return "", err
 		}
 	}
-	if !strings.HasSuffix(host, "."+domain) {
-		return path, nil
+	for _, domain := range domains {
+		if !strings.HasSuffix(host, "."+domain) {
+			continue
+		}
+		bucket := strings.TrimSuffix(host, "."+domain)
+		return slashSeparator + pathJoin(bucket, path), nil
 	}
-	bucket := strings.TrimSuffix(host, "."+domain)
-	return slashSeparator + pathJoin(bucket, path), nil
+	return path, nil
 }

 // If none of the http routes match respond with MethodNotAllowed, in JSON
@@ -213,15 +213,15 @@ func TestGetResource(t *testing.T) {
 	testCases := []struct {
 		p                string
 		host             string
-		domain           string
+		domains          []string
 		expectedResource string
 	}{
-		{"/a/b/c", "test.mydomain.com", "mydomain.com", "/test/a/b/c"},
-		{"/a/b/c", "test.mydomain.com", "notmydomain.com", "/a/b/c"},
-		{"/a/b/c", "test.mydomain.com", "", "/a/b/c"},
+		{"/a/b/c", "test.mydomain.com", []string{"mydomain.com"}, "/test/a/b/c"},
+		{"/a/b/c", "test.mydomain.com", []string{"notmydomain.com"}, "/a/b/c"},
+		{"/a/b/c", "test.mydomain.com", nil, "/a/b/c"},
 	}
 	for i, test := range testCases {
-		gotResource, err := getResource(test.p, test.host, test.domain)
+		gotResource, err := getResource(test.p, test.host, test.domains)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -2094,7 +2094,7 @@ func (api objectAPIHandlers) ListObjectPartsHandler(w http.ResponseWriter, r *ht
 		return
 	}

-	uploadID, partNumberMarker, maxParts, _, s3Error := getObjectResources(r.URL.Query())
+	uploadID, partNumberMarker, maxParts, encodingType, s3Error := getObjectResources(r.URL.Query())
 	if s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
 		return
@@ -2145,34 +2145,59 @@ func (api objectAPIHandlers) ListObjectPartsHandler(w http.ResponseWriter, r *ht
 		}
 	}

-	response := generateListPartsResponse(listPartsInfo)
+	response := generateListPartsResponse(listPartsInfo, encodingType)
 	encodedSuccessResponse := encodeResponse(response)

 	// Write success response.
 	writeSuccessResponseXML(w, encodedSuccessResponse)
 }

-// Send white spaces every 10 seconds to the client till completeMultiPartUpload() is done so that the client does not time out.
-// Downside is we might send 200OK and then send error XML. But accoording to S3 spec
-// the client is supposed to check for error XML even if it received 200OK. But for erasure this is not a
-// problem as completeMultiPartUpload() is quick. Even For FS, it would not be an issue
-// as we do background append as and when the parts arrive and completeMultiPartUpload is quick.
-// Only in a rare case where parts would be out of order will FS' completeMultiPartUpload() take a longer time.
-func sendWhiteSpace(ctx context.Context, w http.ResponseWriter) <-chan struct{} {
-	doneCh := make(chan struct{})
+type whiteSpaceWriter struct {
+	http.ResponseWriter
+	http.Flusher
+	written bool
+}
+
+func (w *whiteSpaceWriter) Write(b []byte) (n int, err error) {
+	n, err = w.ResponseWriter.Write(b)
+	w.written = true
+	return
+}
+
+func (w *whiteSpaceWriter) WriteHeader(statusCode int) {
+	if !w.written {
+		w.ResponseWriter.WriteHeader(statusCode)
+	}
+}
+
+// Send empty whitespaces every 10 seconds to the client till completeMultiPartUpload() is
+// done so that the client does not time out. Downside is we might send 200 OK and
+// then send error XML. But accoording to S3 spec the client is supposed to check
+// for error XML even if it received 200 OK. But for erasure this is not a problem
+// as completeMultiPartUpload() is quick. Even For FS, it would not be an issue as
+// we do background append as and when the parts arrive and completeMultiPartUpload
+// is quick. Only in a rare case where parts would be out of order will
+// FS:completeMultiPartUpload() take a longer time.
+func sendWhiteSpace(ctx context.Context, w http.ResponseWriter) <-chan bool {
+	doneCh := make(chan bool)
 	go func() {
 		ticker := time.NewTicker(time.Second * 10)
+		headerWritten := false
 		for {
 			select {
 			case <-ticker.C:
-				// Write a white space char to the client to prevent client timeouts
-				// when the server takes a long time to completeMultiPartUpload()
-				if _, err := w.Write([]byte("\n\r")); err != nil {
-					logger.LogIf(ctx, err)
-					return
+				// Write header if not written yet.
+				if !headerWritten {
+					w.Write([]byte(xml.Header))
+					headerWritten = true
 				}
+
+				// Once header is written keep writing empty spaces
+				// which are ignored by client SDK XML parsers.
+				// This occurs when server takes long time to completeMultiPartUpload()
+				w.Write([]byte(" "))
 				w.(http.Flusher).Flush()
-			case doneCh <- struct{}{}:
+			case doneCh <- headerWritten:
 				ticker.Stop()
 				return
 			}
@@ -2309,31 +2334,54 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 	if api.CacheAPI() != nil {
 		completeMultiPartUpload = api.CacheAPI().CompleteMultipartUpload
 	}
+
+	// This code is specifically to handle the requirements for slow
+	// complete multipart upload operations on FS mode.
+	writeErrorResponseWithoutXMLHeader := func(ctx context.Context, w http.ResponseWriter, err APIError, reqURL *url.URL) {
+		switch err.Code {
+		case "SlowDown", "XMinioServerNotInitialized", "XMinioReadQuorum", "XMinioWriteQuorum":
+			// Set retry-after header to indicate user-agents to retry request after 120secs.
+			// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Retry-After
+			w.Header().Set("Retry-After", "120")
+		}
+
+		// Generate error response.
+		errorResponse := getAPIErrorResponse(ctx, err, reqURL.Path,
+			w.Header().Get(responseRequestIDKey), w.Header().Get(responseDeploymentIDKey))
+		encodedErrorResponse, _ := xml.Marshal(errorResponse)
+		setCommonHeaders(w)
+		w.Header().Set("Content-Type", string(mimeXML))
+		w.Write(encodedErrorResponse)
+		w.(http.Flusher).Flush()
+	}
+	w = &whiteSpaceWriter{ResponseWriter: w, Flusher: w.(http.Flusher)}
 	completeDoneCh := sendWhiteSpace(ctx, w)
 	objInfo, err := completeMultiPartUpload(ctx, bucket, object, uploadID, completeParts, opts)
 	// Stop writing white spaces to the client. Note that close(doneCh) style is not used as it
 	// can cause white space to be written after we send XML response in a race condition.
-	<-completeDoneCh
+	headerWritten := <-completeDoneCh
 	if err != nil {
-		switch oErr := err.(type) {
-		case PartTooSmall:
-			// Write part too small error.
-			writePartSmallErrorResponse(ctx, w, r, oErr)
-		default:
-			// Handle all other generic issues.
+		if headerWritten {
+			writeErrorResponseWithoutXMLHeader(ctx, w, toAPIError(ctx, err), r.URL)
+		} else {
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 		}
 		return
 	}

 	// Get object location.
-	location := getObjectLocation(r, globalDomainName, bucket, object)
+	location := getObjectLocation(r, globalDomainNames, bucket, object)
 	// Generate complete multipart response.
 	response := generateCompleteMultpartUploadResponse(bucket, object, location, objInfo.ETag)
-	encodedSuccessResponse := encodeResponse(response)
-	if err != nil {
-		writeErrorResponse(ctx, w, toAdminAPIErr(ctx, err), r.URL, guessIsBrowserReq(r))
-		return
+	var encodedSuccessResponse []byte
+	if !headerWritten {
+		encodedSuccessResponse = encodeResponse(response)
+	} else {
+		encodedSuccessResponse, err = xml.Marshal(response)
+		if err != nil {
+			writeErrorResponseWithoutXMLHeader(ctx, w, toAPIError(ctx, err), r.URL)
+			return
+		}
 	}

 	// Set etag.
@@ -2740,9 +2740,9 @@ func testAPICompleteMultipartHandler(obj ObjectLayer, instanceType, bucketName s
 			accessKey: credentials.AccessKey,
 			secretKey: credentials.SecretKey,

-			expectedContent: encodeResponse(completeMultipartAPIError{int64(4), int64(5242880), 1, "e2fc714c4727ee9395f324cd2e7f331f",
-				getAPIErrorResponse(ctx, toAPIError(ctx, PartTooSmall{PartNumber: 1}),
-					getGetObjectURL("", bucketName, objectName), "", "")}),
+			expectedContent: encodeResponse(getAPIErrorResponse(ctx,
+				toAPIError(ctx, PartTooSmall{PartNumber: 1}),
+				getGetObjectURL("", bucketName, objectName), "", "")),
 			expectedRespStatus: http.StatusBadRequest,
 		},
 		// Test case - 5.
@@ -1658,62 +1658,59 @@ func (s *TestSuiteCommon) TestListObjectsHandler(c *check) {
 	c.Assert(err, nil)
 	c.Assert(response.StatusCode, http.StatusOK)

-	buffer1 := bytes.NewReader([]byte("Hello World"))
-	request, err = newTestSignedRequest("PUT", getPutObjectURL(s.endPoint, bucketName, "bar"),
-		int64(buffer1.Len()), buffer1, s.accessKey, s.secretKey, s.signer)
-	c.Assert(err, nil)
+	for _, objectName := range []string{"foo bar 1", "foo bar 2"} {
+		buffer := bytes.NewReader([]byte("Hello World"))
+		request, err = newTestSignedRequest("PUT", getPutObjectURL(s.endPoint, bucketName, objectName),
+			int64(buffer.Len()), buffer, s.accessKey, s.secretKey, s.signer)
+		c.Assert(err, nil)

-	client = http.Client{Transport: s.transport}
-	response, err = client.Do(request)
-	c.Assert(err, nil)
-	c.Assert(response.StatusCode, http.StatusOK)
+		client = http.Client{Transport: s.transport}
+		response, err = client.Do(request)
+		c.Assert(err, nil)
+		c.Assert(response.StatusCode, http.StatusOK)
+	}

-	// create listObjectsV1 request with valid parameters
-	request, err = newTestSignedRequest("GET", getListObjectsV1URL(s.endPoint, bucketName, "1000"),
-		0, nil, s.accessKey, s.secretKey, s.signer)
-	c.Assert(err, nil)
-	client = http.Client{Transport: s.transport}
-	// execute the HTTP request.
-	response, err = client.Do(request)
-	c.Assert(err, nil)
-	c.Assert(response.StatusCode, http.StatusOK)
+	var testCases = []struct {
+		getURL          string
+		expectedStrings []string
+	}{
+		{getListObjectsV1URL(s.endPoint, bucketName, "", "1000", ""), []string{"<Key>foo bar 1</Key>", "<Key>foo bar 2</Key>"}},
+		{getListObjectsV1URL(s.endPoint, bucketName, "", "1000", "url"), []string{"<Key>foo+bar+1</Key>", "<Key>foo+bar+2</Key>"}},
+		{getListObjectsV2URL(s.endPoint, bucketName, "", "1000", "", ""),
+			[]string{
+				"<Key>foo bar 1</Key>",
+				"<Key>foo bar 2</Key>",
+				"<Owner><ID></ID><DisplayName></DisplayName></Owner>",
+			},
+		},
+		{getListObjectsV2URL(s.endPoint, bucketName, "", "1000", "true", ""),
+			[]string{
+				"<Key>foo bar 1</Key>",
+				"<Key>foo bar 2</Key>",
+				fmt.Sprintf("<Owner><ID>%s</ID><DisplayName></DisplayName></Owner>", globalMinioDefaultOwnerID),
+			},
+		},
+		{getListObjectsV2URL(s.endPoint, bucketName, "", "1000", "", "url"), []string{"<Key>foo+bar+1</Key>", "<Key>foo+bar+2</Key>"}},
+	}

-	getContent, err := ioutil.ReadAll(response.Body)
-	c.Assert(err, nil)
-	c.Assert(strings.Contains(string(getContent), "<Key>bar</Key>"), true)
+	for i, testCase := range testCases {
+		// create listObjectsV1 request with valid parameters
+		request, err = newTestSignedRequest("GET", testCase.getURL, 0, nil, s.accessKey, s.secretKey, s.signer)
+		c.Assert(err, nil)
+		client = http.Client{Transport: s.transport}
+		// execute the HTTP request.
+		response, err = client.Do(request)
+		c.Assert(err, nil)
+		c.Assert(response.StatusCode, http.StatusOK)

-	// create listObjectsV2 request with valid parameters
-	request, err = newTestSignedRequest("GET", getListObjectsV2URL(s.endPoint, bucketName, "1000", ""),
-		0, nil, s.accessKey, s.secretKey, s.signer)
-	c.Assert(err, nil)
-	client = http.Client{Transport: s.transport}
-	// execute the HTTP request.
-	response, err = client.Do(request)
-	c.Assert(err, nil)
-	c.Assert(response.StatusCode, http.StatusOK)
-
-	getContent, err = ioutil.ReadAll(response.Body)
-	c.Assert(err, nil)
-	c.Assert(strings.Contains(string(getContent), "<Key>bar</Key>"), true)
-	c.Assert(strings.Contains(string(getContent), "<Owner><ID></ID><DisplayName></DisplayName></Owner>"), true)
-
-	// create listObjectsV2 request with valid parameters and fetch-owner activated
-	request, err = newTestSignedRequest("GET", getListObjectsV2URL(s.endPoint, bucketName, "1000", "true"),
-		0, nil, s.accessKey, s.secretKey, s.signer)
-	c.Assert(err, nil)
-	client = http.Client{Transport: s.transport}
-	// execute the HTTP request.
-	response, err = client.Do(request)
-	c.Assert(err, nil)
-	c.Assert(response.StatusCode, http.StatusOK)
-
-	getContent, err = ioutil.ReadAll(response.Body)
-	c.Assert(err, nil)
-
-	c.Assert(strings.Contains(string(getContent), "<Key>bar</Key>"), true)
-	c.Assert(strings.Contains(string(getContent), fmt.Sprintf("<Owner><ID>%s</ID><DisplayName></DisplayName></Owner>",
-		globalMinioDefaultOwnerID)), true)
+		getContent, err := ioutil.ReadAll(response.Body)
+		c.Assert(err, nil)

+		fmt.Printf("Test %d: %+v vs %+v\n", i+1, string(getContent), testCase.expectedStrings)
+		for _, expectedStr := range testCase.expectedStrings {
+			c.Assert(strings.Contains(string(getContent), expectedStr), true)
+		}
+	}
 }

 // TestListObjectsHandlerErrors - Setting invalid parameters to List Objects
@@ -1733,7 +1730,7 @@ func (s *TestSuiteCommon) TestListObjectsHandlerErrors(c *check) {
 	c.Assert(response.StatusCode, http.StatusOK)

 	// create listObjectsV1 request with invalid value of max-keys parameter. max-keys is set to -2.
-	request, err = newTestSignedRequest("GET", getListObjectsV1URL(s.endPoint, bucketName, "-2"),
+	request, err = newTestSignedRequest("GET", getListObjectsV1URL(s.endPoint, bucketName, "", "-2", ""),
 		0, nil, s.accessKey, s.secretKey, s.signer)
 	c.Assert(err, nil)
 	client = http.Client{Transport: s.transport}
@@ -1744,7 +1741,7 @@ func (s *TestSuiteCommon) TestListObjectsHandlerErrors(c *check) {
 	verifyError(c, response, "InvalidArgument", "Argument maxKeys must be an integer between 0 and 2147483647", http.StatusBadRequest)

 	// create listObjectsV2 request with invalid value of max-keys parameter. max-keys is set to -2.
-	request, err = newTestSignedRequest("GET", getListObjectsV2URL(s.endPoint, bucketName, "-2", ""),
+	request, err = newTestSignedRequest("GET", getListObjectsV2URL(s.endPoint, bucketName, "", "-2", "", ""),
 		0, nil, s.accessKey, s.secretKey, s.signer)
 	c.Assert(err, nil)
 	client = http.Client{Transport: s.transport}
@@ -163,7 +163,7 @@ func doesPresignV2SignatureMatch(r *http.Request) APIErrorCode {
 		return ErrExpiredPresignRequest
 	}

-	encodedResource, err = getResource(encodedResource, r.Host, globalDomainName)
+	encodedResource, err = getResource(encodedResource, r.Host, globalDomainNames)
 	if err != nil {
 		return ErrInvalidRequest
 	}
@@ -257,7 +257,7 @@ func doesSignV2Match(r *http.Request) APIErrorCode {
 		return ErrInvalidQueryParams
 	}

-	encodedResource, err = getResource(encodedResource, r.Host, globalDomainName)
+	encodedResource, err = getResource(encodedResource, r.Host, globalDomainNames)
 	if err != nil {
 		return ErrInvalidRequest
 	}
@@ -1458,16 +1458,19 @@ func getBucketLocationURL(endPoint, bucketName string) string {
 }

 // return URL for listing objects in the bucket with V1 legacy API.
-func getListObjectsV1URL(endPoint, bucketName string, maxKeys string) string {
+func getListObjectsV1URL(endPoint, bucketName, prefix, maxKeys, encodingType string) string {
 	queryValue := url.Values{}
 	if maxKeys != "" {
 		queryValue.Set("max-keys", maxKeys)
 	}
-	return makeTestTargetURL(endPoint, bucketName, "", queryValue)
+	if encodingType != "" {
+		queryValue.Set("encoding-type", encodingType)
+	}
+	return makeTestTargetURL(endPoint, bucketName, prefix, queryValue)
 }

 // return URL for listing objects in the bucket with V2 API.
-func getListObjectsV2URL(endPoint, bucketName string, maxKeys string, fetchOwner string) string {
+func getListObjectsV2URL(endPoint, bucketName, prefix, maxKeys, fetchOwner, encodingType string) string {
 	queryValue := url.Values{}
 	queryValue.Set("list-type", "2") // Enables list objects V2 URL.
 	if maxKeys != "" {
@@ -1476,7 +1479,10 @@ func getListObjectsV2URL(endPoint, bucketName string, maxKeys string, fetchOwner
 	if fetchOwner != "" {
 		queryValue.Set("fetch-owner", fetchOwner)
 	}
-	return makeTestTargetURL(endPoint, bucketName, "", queryValue)
+	if encodingType != "" {
+		queryValue.Set("encoding-type", encodingType)
+	}
+	return makeTestTargetURL(endPoint, bucketName, prefix, queryValue)
 }

 // return URL for a new multipart upload.
@@ -83,6 +83,11 @@ func findEndpoint() string {
 			if strings.Count(addr, ":") > 0 {
 				addr = strings.Join([]string{"[", addr, "]"}, "")
 			}
+			// wait for cmd to complete before return
+			if err = cmd.Wait(); err != nil {
+				// command failed to run
+				log.Fatal(err.Error())
+			}
 			// return joint address and port
 			return strings.Join([]string{addr, port}, ":")
 		}
@@ -163,6 +163,7 @@ The Minio server configuration file is stored on the backend in json format. The
 | `username` | _string_ | Username to connect to the MQTT server (if required) |
 | `password` | _string_ | Password to connect to the MQTT server (if required) |
 | `queueDir` | _string_ | Persistent store for events when MQTT broker is offline |
+| `queueLimit` | _int_ | Set the maximum event limit for the persistent store. The default limit is 10000 |

 An example configuration for MQTT is shown below:

@@ -175,12 +176,12 @@ An example configuration for MQTT is shown below:
        "qos": 1,
        "username": "",
        "password": "",
-        "queueDir": ""
+        "queueDir": "",
+        "queueLimit": 0
    }
 }
 ```
-MQTT supports persistent event store in the client-side. The persistent store will backup events when the MQTT broker goes offline and replays it when the broker comes back online. The event store can be configured by setting the directory path in `queueDir` field in the mqtt config. For eg, the `queueDir` can be `/home/events`.
-
+Minio supports persistent event store. The persistent store will backup events when the MQTT broker goes offline and replays it when the broker comes back online. The event store can be configured by setting the directory path in `queueDir` field and the maximum limit of events in the queueDir in `queueLimit` field. For eg, the `queueDir` can be `/home/events` and `queueLimit` can be `1000`. By default, the `queueLimit` is set to 10000.

 To update the configuration, use `mc admin config get` command to get the current configuration file for the minio deployment in json format, and save it locally.
 ```sh
@@ -549,7 +550,7 @@ Minio server also supports [NATS Streaming mode](http://nats.io/documentation/st
        "password": "yoursecret",
        "token": "",
        "secure": false,
-        "pingInterval": 0
+        "pingInterval": 0,
        "streaming": {
            "enable": true,
            "clusterID": "test-cluster",
@@ -156,6 +156,12 @@ export MINIO_DOMAIN=mydomain.com
 minio server /data
 ```

+For advanced use cases `MINIO_DOMAIN` environment variable supports multiple-domains with comma separated values.
+```sh
+export MINIO_DOMAIN=sub1.mydomain.com,sub2.mydomain.com
+minio server /data
+```
+
 ### Drive Sync

 By default, Minio writes to disk in synchronous mode for all metadata operations. Set `MINIO_DRIVE_SYNC` environment variable to enable synchronous mode for all data operations as well.
@@ -86,7 +86,8 @@
 				"password": "",
 				"reconnectInterval": 0,
 				"keepAliveInterval": 0,
-				"queueDir": ""
+				"queueDir": "",
+                                "queueLimit": 0
 			}
 		},
 		"mysql": {
@@ -5,7 +5,7 @@ version: '2'
 # 9001 through 9004.
 services:
 minio1:
-  image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+  image: minio/minio:RELEASE.2019-02-20T22-44-29Z
  volumes:
   - data1:/data
  ports:
@@ -15,7 +15,7 @@ services:
   MINIO_SECRET_KEY: minio123
  command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
 minio2:
-  image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+  image: minio/minio:RELEASE.2019-02-20T22-44-29Z
  volumes:
   - data2:/data
  ports:
@@ -25,7 +25,7 @@ services:
   MINIO_SECRET_KEY: minio123
  command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
 minio3:
-  image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+  image: minio/minio:RELEASE.2019-02-20T22-44-29Z
  volumes:
   - data3:/data
  ports:
@@ -35,7 +35,7 @@ services:
   MINIO_SECRET_KEY: minio123
  command: server http://minio1/data http://minio2/data http://minio3/data http://minio4/data 
 minio4:
-  image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+  image: minio/minio:RELEASE.2019-02-20T22-44-29Z
  volumes:
   - data4:/data
  ports:
@@ -2,7 +2,7 @@ version: '3.1'

 services:
  minio1:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio1
    volumes:
      - minio1-data:/export
@@ -24,7 +24,7 @@ services:
      - access_key

  minio2:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio2
    volumes:
      - minio2-data:/export
@@ -46,7 +46,7 @@ services:
      - access_key

  minio3:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio3
    volumes:
      - minio3-data:/export
@@ -68,7 +68,7 @@ services:
      - access_key

  minio4:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio4
    volumes:
      - minio4-data:/export
@@ -2,7 +2,7 @@ version: '3'

 services:
  minio1:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio1
    volumes:
      - minio1-data:/export
@@ -24,7 +24,7 @@ services:
    command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export

  minio2:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio2
    volumes:
      - minio2-data:/export
@@ -46,7 +46,7 @@ services:
    command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export

  minio3:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio3
    volumes:
      - minio3-data:/export
@@ -68,7 +68,7 @@ services:
    command: server http://minio1/export http://minio2/export http://minio3/export http://minio4/export

  minio4:
-    image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+    image: minio/minio:RELEASE.2019-02-20T22-44-29Z
    hostname: minio4
    volumes:
      - minio4-data:/export
@@ -30,7 +30,7 @@ spec:
          value: "minio"
        - name: MINIO_SECRET_KEY
          value: "minio123"
-        image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+        image: minio/minio:RELEASE.2019-02-20T22-44-29Z
        # Unfortunately you must manually define each server. Perhaps autodiscovery via DNS can be implemented in the future.
        args:
        - server
@@ -21,7 +21,7 @@ spec:
          value: "minio"
        - name: MINIO_SECRET_KEY
          value: "minio123"
-        image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+        image: minio/minio:RELEASE.2019-02-20T22-44-29Z
        args:
        - server
        - http://minio-0.minio.default.svc.cluster.local/data
@@ -21,7 +21,7 @@ spec:
      containers:
      - name: minio
        # Pulls the default Minio image from Docker Hub
-        image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+        image: minio/minio:RELEASE.2019-02-20T22-44-29Z
        args:
        - gateway
        - gcs
@@ -29,7 +29,7 @@ spec:
        - name: data 
          mountPath: "/data"
        # Pulls the lastest Minio image from Docker Hub
-        image: minio/minio:RELEASE.2019-02-14T00-21-45Z
+        image: minio/minio:RELEASE.2019-02-20T22-44-29Z
        args:
        - server
        - /data
@@ -47,14 +47,30 @@ func newCoreDNSMsg(bucket, ip string, port int, ttl uint32) ([]byte, error) {

 // Retrieves list of DNS entries for the domain.
 func (c *coreDNS) List() ([]SrvRecord, error) {
-	key := msg.Path(fmt.Sprintf("%s.", c.domainName), defaultPrefixPath)
-	return c.list(key)
+	var srvRecords []SrvRecord
+	for _, domainName := range c.domainNames {
+		key := msg.Path(fmt.Sprintf("%s.", domainName), defaultPrefixPath)
+		records, err := c.list(key)
+		if err != nil {
+			return nil, err
+		}
+		srvRecords = append(srvRecords, records...)
+	}
+	return srvRecords, nil
 }

 // Retrieves DNS records for a bucket.
 func (c *coreDNS) Get(bucket string) ([]SrvRecord, error) {
-	key := msg.Path(fmt.Sprintf("%s.%s.", bucket, c.domainName), defaultPrefixPath)
-	return c.list(key)
+	var srvRecords []SrvRecord
+	for _, domainName := range c.domainNames {
+		key := msg.Path(fmt.Sprintf("%s.%s.", bucket, domainName), defaultPrefixPath)
+		records, err := c.list(key)
+		if err != nil {
+			return nil, err
+		}
+		srvRecords = append(srvRecords, records...)
+	}
+	return srvRecords, nil
 }

 // Retrieves list of entries under the key passed.
@@ -92,12 +108,14 @@ func (c *coreDNS) list(key string) ([]SrvRecord, error) {
 		//
 		// In all other situations when we want to list all DNS records,
 		// which is handled in the else clause.
-		if key != msg.Path(fmt.Sprintf(".%s.", c.domainName), defaultPrefixPath) {
-			if srvRecord.Key == "/" {
+		for _, domainName := range c.domainNames {
+			if key != msg.Path(fmt.Sprintf(".%s.", domainName), defaultPrefixPath) {
+				if srvRecord.Key == "/" {
+					srvRecords = append(srvRecords, srvRecord)
+				}
+			} else {
 				srvRecords = append(srvRecords, srvRecord)
 			}
-		} else {
-			srvRecords = append(srvRecords, srvRecord)
 		}

 	}
@@ -117,16 +135,18 @@ func (c *coreDNS) Put(bucket string) error {
 		if err != nil {
 			return err
 		}
-		key := msg.Path(fmt.Sprintf("%s.%s", bucket, c.domainName), defaultPrefixPath)
-		key = key + "/" + ip
-		ctx, cancel := context.WithTimeout(context.Background(), defaultContextTimeout)
-		_, err = c.etcdClient.Put(ctx, key, string(bucketMsg))
-		defer cancel()
-		if err != nil {
-			ctx, cancel = context.WithTimeout(context.Background(), defaultContextTimeout)
-			c.etcdClient.Delete(ctx, key)
+		for _, domainName := range c.domainNames {
+			key := msg.Path(fmt.Sprintf("%s.%s", bucket, domainName), defaultPrefixPath)
+			key = key + "/" + ip
+			ctx, cancel := context.WithTimeout(context.Background(), defaultContextTimeout)
+			_, err = c.etcdClient.Put(ctx, key, string(bucketMsg))
 			defer cancel()
-			return err
+			if err != nil {
+				ctx, cancel = context.WithTimeout(context.Background(), defaultContextTimeout)
+				c.etcdClient.Delete(ctx, key)
+				defer cancel()
+				return err
+			}
 		}
 	}
 	return nil
@@ -134,33 +154,35 @@ func (c *coreDNS) Put(bucket string) error {

 // Removes DNS entries added in Put().
 func (c *coreDNS) Delete(bucket string) error {
-	key := msg.Path(fmt.Sprintf("%s.%s.", bucket, c.domainName), defaultPrefixPath)
-	srvRecords, err := c.list(key)
-	if err != nil {
-		return err
-	}
-	for _, record := range srvRecords {
-		dctx, dcancel := context.WithTimeout(context.Background(), defaultContextTimeout)
-		if _, err = c.etcdClient.Delete(dctx, key+"/"+record.Host); err != nil {
-			dcancel()
+	for _, domainName := range c.domainNames {
+		key := msg.Path(fmt.Sprintf("%s.%s.", bucket, domainName), defaultPrefixPath)
+		srvRecords, err := c.list(key)
+		if err != nil {
 			return err
 		}
-		dcancel()
+		for _, record := range srvRecords {
+			dctx, dcancel := context.WithTimeout(context.Background(), defaultContextTimeout)
+			if _, err = c.etcdClient.Delete(dctx, key+"/"+record.Host); err != nil {
+				dcancel()
+				return err
+			}
+			dcancel()
+		}
 	}
-	return err
+	return nil
 }

 // CoreDNS - represents dns config for coredns server.
 type coreDNS struct {
-	domainName string
-	domainIPs  set.StringSet
-	domainPort int
-	etcdClient *etcd.Client
+	domainNames []string
+	domainIPs   set.StringSet
+	domainPort  int
+	etcdClient  *etcd.Client
 }

 // NewCoreDNS - initialize a new coreDNS set/unset values.
-func NewCoreDNS(domainName string, domainIPs set.StringSet, domainPort string, etcdClient *etcd.Client) (Config, error) {
-	if domainName == "" || domainIPs.IsEmpty() {
+func NewCoreDNS(domainNames []string, domainIPs set.StringSet, domainPort string, etcdClient *etcd.Client) (Config, error) {
+	if len(domainNames) == 0 || domainIPs.IsEmpty() {
 		return nil, errors.New("invalid argument")
 	}

@@ -170,9 +192,9 @@ func NewCoreDNS(domainName string, domainIPs set.StringSet, domainPort string, e
 	}

 	return &coreDNS{
-		domainName: domainName,
-		domainIPs:  domainIPs,
-		domainPort: port,
-		etcdClient: etcdClient,
+		domainNames: domainNames,
+		domainIPs:   domainIPs,
+		domainPort:  port,
+		etcdClient:  etcdClient,
 	}, nil
 }
@@ -0,0 +1,104 @@
+/*
+ * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ *
+ * 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://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 target
+
+import (
+	"sync"
+
+	"github.com/minio/minio/pkg/event"
+)
+
+const (
+	maxStoreLimit = 10000
+)
+
+// MemoryStore persists events in memory.
+type MemoryStore struct {
+	sync.RWMutex
+	events map[string]event.Event
+	eC     uint16
+	limit  uint16
+}
+
+// NewMemoryStore creates a memory store instance.
+func NewMemoryStore(limit uint16) *MemoryStore {
+	if limit == 0 || limit > maxStoreLimit {
+		limit = maxStoreLimit
+	}
+	memoryStore := &MemoryStore{
+		events: make(map[string]event.Event),
+		limit:  limit,
+	}
+	return memoryStore
+}
+
+// Open is in-effective here.
+// Implemented for interface compatibility.
+func (store *MemoryStore) Open() error {
+	return nil
+}
+
+// Put - puts the event in store.
+func (store *MemoryStore) Put(e event.Event) error {
+	store.Lock()
+	defer store.Unlock()
+	if store.eC == store.limit {
+		return ErrLimitExceeded
+	}
+	key, kErr := getNewUUID()
+	if kErr != nil {
+		return kErr
+	}
+	store.events[key] = e
+	store.eC++
+	return nil
+}
+
+// Get - retrieves the event from store.
+func (store *MemoryStore) Get(key string) (event.Event, error) {
+	store.RLock()
+	defer store.RUnlock()
+
+	if event, exist := store.events[key]; exist {
+		return event, nil
+	}
+
+	return event.Event{}, ErrNoSuchKey
+}
+
+// Del - deletes the event from store.
+func (store *MemoryStore) Del(key string) {
+	store.Lock()
+	defer store.Unlock()
+
+	delete(store.events, key)
+
+	store.eC--
+}
+
+// ListAll - lists all the keys in the store.
+func (store *MemoryStore) ListAll() []string {
+	store.RLock()
+	defer store.RUnlock()
+
+	keys := []string{}
+	for k := range store.events {
+		keys = append(keys, k)
+	}
+
+	return keys
+}
@@ -0,0 +1,106 @@
+/*
+ * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ *
+ * 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://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 target
+
+import (
+	"reflect"
+	"testing"
+)
+
+// TestMemoryStorePut - Tests for store.Put
+func TestMemoryStorePut(t *testing.T) {
+	store := NewMemoryStore(1000)
+	defer func() {
+		store = nil
+	}()
+	for i := 0; i < 100; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	if len(store.ListAll()) != 100 {
+		t.Fatalf("ListAll() Expected: 100, got %d", len(store.ListAll()))
+	}
+}
+
+// TestMemoryStoreGet - Tests for store.Get.
+func TestMemoryStoreGet(t *testing.T) {
+	store := NewMemoryStore(1000)
+	defer func() {
+		store = nil
+	}()
+	for i := 0; i < 10; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	eventKeys := store.ListAll()
+	if len(eventKeys) == 10 {
+		for _, key := range eventKeys {
+			event, eErr := store.Get(key)
+			if eErr != nil {
+				t.Fatal("Failed to Get the event from the queue store ", eErr)
+			}
+			if !reflect.DeepEqual(testEvent, event) {
+				t.Fatalf("Failed to read the event: error: expected = %v, got = %v", testEvent, event)
+			}
+		}
+	} else {
+		t.Fatalf("ListAll() Expected: 10, got %d", len(eventKeys))
+	}
+}
+
+// TestMemoryStoreDel - Tests for store.Del.
+func TestMemoryStoreDel(t *testing.T) {
+	store := NewMemoryStore(1000)
+	defer func() {
+		store = nil
+	}()
+	for i := 0; i < 20; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	eventKeys := store.ListAll()
+	if len(eventKeys) == 20 {
+		for _, key := range eventKeys {
+			store.Del(key)
+		}
+	} else {
+		t.Fatalf("ListAll() Expected: 20, got %d", len(eventKeys))
+	}
+
+	if len(store.ListAll()) != 0 {
+		t.Fatalf("ListAll() Expected: 0, got %d", len(store.ListAll()))
+	}
+}
+
+// TestMemoryStoreLimit - tests for store limit.
+func TestMemoryStoreLimit(t *testing.T) {
+	store := NewMemoryStore(5)
+	defer func() {
+		store = nil
+	}()
+	for i := 0; i < 5; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	if err := store.Put(testEvent); err == nil {
+		t.Fatalf("Expected to fail with %s, but passes", ErrLimitExceeded)
+	}
+}
@@ -30,6 +30,11 @@ import (
 	xnet "github.com/minio/minio/pkg/net"
 )

+const (
+	retryInterval     = 3 // In Seconds
+	reconnectInterval = 5 // In Seconds
+)
+
 // MQTTArgs - MQTT target arguments.
 type MQTTArgs struct {
 	Enable               bool           `json:"enable"`
@@ -42,6 +47,7 @@ type MQTTArgs struct {
 	KeepAlive            time.Duration  `json:"keepAliveInterval"`
 	RootCAs              *x509.CertPool `json:"-"`
 	QueueDir             string         `json:"queueDir"`
+	QueueLimit           uint16         `json:"queueLimit"`
 }

 // Validate MQTTArgs fields
@@ -75,6 +81,8 @@ type MQTTTarget struct {
 	id     event.TargetID
 	args   MQTTArgs
 	client mqtt.Client
+	store  Store
+	reconn bool
 }

 // ID - returns target ID.
@@ -82,16 +90,33 @@ func (target *MQTTTarget) ID() event.TargetID {
 	return target.id
 }

-// Send - sends event to MQTT.
-func (target *MQTTTarget) Send(eventData event.Event) error {
-	if !target.client.IsConnected() {
-		token := target.client.Connect()
-		if token.Wait() {
-			if err := token.Error(); err != nil {
-				return err
+// Reads persisted events from the store and re-plays.
+func (target *MQTTTarget) retry() {
+	target.reconn = true
+	events := target.store.ListAll()
+	for len(events) != 0 {
+		for _, key := range events {
+			event, eErr := target.store.Get(key)
+			if eErr != nil {
+				continue
 			}
+			for !target.client.IsConnectionOpen() {
+				time.Sleep(retryInterval * time.Second)
+			}
+			// The connection is open.
+			if err := target.send(event); err != nil {
+				continue
+			}
+			// Delete after a successful publish.
+			target.store.Del(key)
 		}
+		events = target.store.ListAll()
 	}
+	// Release the reconn state.
+	target.reconn = false
+}
+
+func (target *MQTTTarget) send(eventData event.Event) error {

 	objectName, err := url.QueryUnescape(eventData.S3.Object.Key)
 	if err != nil {
@@ -105,16 +130,27 @@ func (target *MQTTTarget) Send(eventData event.Event) error {
 	}

 	token := target.client.Publish(target.args.Topic, target.args.QoS, false, string(data))
+	token.Wait()
+	return token.Error()

-	// In-case if the queueDir is configured OR mqtt broker is offline, the token.Wait() waits indefinitely
-	if target.args.QueueDir == "" {
-		token.Wait()
-		return token.Error()
+}
+
+// Send - sends event to MQTT when the connection is active.
+func (target *MQTTTarget) Send(eventData event.Event) error {
+	// Persist the events if the connection is not active.
+	if !target.client.IsConnectionOpen() {
+		if err := target.store.Put(eventData); err != nil {
+			return err
+		}
+		// Ignore if retry is triggered already.
+		if !target.reconn {
+			go target.retry()
+		}
+		return nil
 	}

-	// Need a fix from the paho.mqtt.golang library - https://github.com/eclipse/paho.mqtt.golang/issues/274
-	// Right now the server panics on IO File store errors, returning nil for now.
-	return nil
+	// Publishes to the broker as the connection is active.
+	return target.send(eventData)
 }

 // Close - does nothing and available for interface compatibility.
@@ -134,19 +170,49 @@ func NewMQTTTarget(id string, args MQTTArgs) (*MQTTTarget, error) {
 		SetTLSConfig(&tls.Config{RootCAs: args.RootCAs}).
 		AddBroker(args.Broker.String())

+	var store Store
+
 	if args.QueueDir != "" {
-		options = options.SetStore(mqtt.NewFileStore(args.QueueDir))
+		store = NewQueueStore(args.QueueDir, args.QueueLimit)
+		if oErr := store.Open(); oErr != nil {
+			return nil, oErr
+		}
+	} else {
+		store = NewMemoryStore(args.QueueLimit)
 	}

 	client := mqtt.NewClient(options)
-	token := client.Connect()
-	if token.Wait() && token.Error() != nil {
-		return nil, token.Error()
-	}

-	return &MQTTTarget{
+	// The client should establish a first time connection.
+	// Connect() should be successful atleast once to publish events.
+	token := client.Connect()
+
+	go func() {
+		// Repeat the pings until the client registers the clientId and receives a token.
+		for {
+			if token.Wait() && token.Error() == nil {
+				// Connected
+				break
+			}
+			// Reconnecting
+			time.Sleep(reconnectInterval * time.Second)
+			token = client.Connect()
+		}
+	}()
+
+	target := &MQTTTarget{
 		id:     event.TargetID{ID: id, Name: "mqtt"},
 		args:   args,
 		client: client,
-	}, nil
+		store:  store,
+		reconn: false,
+	}
+
+	// Replay any previously persisted events in the store.
+	if len(target.store.ListAll()) != 0 {
+		go target.retry()
+
+	}
+
+	return target, nil
 }
@@ -0,0 +1,178 @@
+/*
+ * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ *
+ * 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://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 target
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+
+	"github.com/minio/minio/pkg/event"
+)
+
+const (
+	maxLimit = 10000 // Max store limit.
+	eventExt = ".event"
+)
+
+// QueueStore - Filestore for persisting events.
+type QueueStore struct {
+	sync.RWMutex
+	directory string
+	eC        uint16
+	limit     uint16
+}
+
+// NewQueueStore - Creates an instance for QueueStore.
+func NewQueueStore(directory string, limit uint16) *QueueStore {
+	if limit == 0 {
+		limit = maxLimit
+	}
+	queueStore := &QueueStore{
+		directory: directory,
+		limit:     limit,
+	}
+	return queueStore
+}
+
+// Open - Creates the directory if not present.
+func (store *QueueStore) Open() error {
+	store.Lock()
+	defer store.Unlock()
+
+	if terr := os.MkdirAll(store.directory, os.FileMode(0770)); terr != nil {
+		return terr
+	}
+
+	eCount := uint16(len(store.listAll()))
+	if eCount >= store.limit {
+		return ErrLimitExceeded
+	}
+
+	store.eC = eCount
+
+	return nil
+}
+
+// write - writes event to the directory.
+func (store *QueueStore) write(directory string, key string, e event.Event) error {
+
+	// Marshalls the event.
+	eventData, err := json.Marshal(e)
+	if err != nil {
+		return err
+	}
+
+	path := filepath.Join(store.directory, key+eventExt)
+	if err := ioutil.WriteFile(path, eventData, os.FileMode(0770)); err != nil {
+		return err
+	}
+
+	// Increment the event count.
+	store.eC++
+
+	return nil
+}
+
+// Put - puts a event to the store.
+func (store *QueueStore) Put(e event.Event) error {
+	store.Lock()
+	defer store.Unlock()
+	if store.eC >= store.limit {
+		return ErrLimitExceeded
+	}
+	key, kErr := getNewUUID()
+	if kErr != nil {
+		return kErr
+	}
+	return store.write(store.directory, key, e)
+}
+
+// Get - gets a event from the store.
+func (store *QueueStore) Get(key string) (event.Event, error) {
+	store.RLock()
+	defer store.RUnlock()
+
+	var event event.Event
+
+	filepath := filepath.Join(store.directory, key+eventExt)
+
+	eventData, rerr := ioutil.ReadFile(filepath)
+	if rerr != nil {
+		store.del(key)
+		return event, rerr
+	}
+
+	if len(eventData) == 0 {
+		store.del(key)
+	}
+
+	uerr := json.Unmarshal(eventData, &event)
+	if uerr != nil {
+		store.del(key)
+		return event, uerr
+	}
+
+	return event, nil
+}
+
+// Del - Deletes an entry from the store.
+func (store *QueueStore) Del(key string) {
+	store.Lock()
+	defer store.Unlock()
+	store.del(key)
+}
+
+// lockless call
+func (store *QueueStore) del(key string) {
+	p := filepath.Join(store.directory, key+eventExt)
+
+	rerr := os.Remove(p)
+	if rerr != nil {
+		return
+	}
+
+	// Decrement the event count.
+	store.eC--
+}
+
+// ListAll - lists all the keys in the directory.
+func (store *QueueStore) ListAll() []string {
+	store.RLock()
+	defer store.RUnlock()
+	return store.listAll()
+}
+
+// lockless call.
+func (store *QueueStore) listAll() []string {
+	var err error
+	var keys []string
+	var files []os.FileInfo
+
+	files, err = ioutil.ReadDir(store.directory)
+	if err != nil {
+		return nil
+	}
+
+	for _, f := range files {
+		keys = append(keys, strings.TrimSuffix(f.Name(), eventExt))
+	}
+	return keys
+}
@@ -0,0 +1,162 @@
+/*
+ * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ *
+ * 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://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 target
+
+import (
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+
+	"github.com/minio/minio/pkg/event"
+)
+
+// TestDir
+var queueDir = filepath.Join(os.TempDir(), "minio_test")
+
+// Sample test event.
+var testEvent = event.Event{EventVersion: "1.0", EventSource: "test_source", AwsRegion: "test_region", EventTime: "test_time", EventName: event.ObjectAccessedGet}
+
+// Initialize the store.
+func setUpStore(directory string, limit uint16) (Store, error) {
+	store := NewQueueStore(queueDir, limit)
+	if oErr := store.Open(); oErr != nil {
+		return nil, oErr
+	}
+	return store, nil
+}
+
+// Tear down store
+func tearDownStore() error {
+	if err := os.RemoveAll(queueDir); err != nil {
+		return err
+	}
+	return nil
+}
+
+// TestQueueStorePut - tests for store.Put
+func TestQueueStorePut(t *testing.T) {
+	defer func() {
+		if err := tearDownStore(); err != nil {
+			t.Fatal("Failed to tear down store ", err)
+		}
+	}()
+	store, err := setUpStore(queueDir, 10000)
+	if err != nil {
+		t.Fatal("Failed to create a queue store ", err)
+
+	}
+	// Put 100 events.
+	for i := 0; i < 100; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	// Count the events.
+	if len(store.ListAll()) != 100 {
+		t.Fatalf("ListAll() Expected: 100, got %d", len(store.ListAll()))
+	}
+}
+
+// TestQueueStoreGet - tests for store.Get
+func TestQueueStoreGet(t *testing.T) {
+	defer func() {
+		if err := tearDownStore(); err != nil {
+			t.Fatal("Failed to tear down store ", err)
+		}
+	}()
+	store, err := setUpStore(queueDir, 10000)
+	if err != nil {
+		t.Fatal("Failed to create a queue store ", err)
+	}
+	// Put 10 events
+	for i := 0; i < 10; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	eventKeys := store.ListAll()
+	// Get 10 events.
+	if len(eventKeys) == 10 {
+		for _, key := range eventKeys {
+			event, eErr := store.Get(key)
+			if eErr != nil {
+				t.Fatal("Failed to Get the event from the queue store ", eErr)
+			}
+			if !reflect.DeepEqual(testEvent, event) {
+				t.Fatalf("Failed to read the event: error: expected = %v, got = %v", testEvent, event)
+			}
+		}
+	} else {
+		t.Fatalf("ListAll() Expected: 10, got %d", len(eventKeys))
+	}
+}
+
+// TestQueueStoreDel - tests for store.Del
+func TestQueueStoreDel(t *testing.T) {
+	defer func() {
+		if err := tearDownStore(); err != nil {
+			t.Fatal("Failed to tear down store ", err)
+		}
+	}()
+	store, err := setUpStore(queueDir, 10000)
+	if err != nil {
+		t.Fatal("Failed to create a queue store ", err)
+	}
+	// Put 20 events.
+	for i := 0; i < 20; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	eventKeys := store.ListAll()
+	// Remove all the events.
+	if len(eventKeys) == 20 {
+		for _, key := range eventKeys {
+			store.Del(key)
+		}
+	} else {
+		t.Fatalf("ListAll() Expected: 20, got %d", len(eventKeys))
+	}
+
+	if len(store.ListAll()) != 0 {
+		t.Fatalf("ListAll() Expected: 0, got %d", len(store.ListAll()))
+	}
+}
+
+// TestQueueStoreLimit - tests the event limit for the store.
+func TestQueueStoreLimit(t *testing.T) {
+	defer func() {
+		if err := tearDownStore(); err != nil {
+			t.Fatal("Failed to tear down store ", err)
+		}
+	}()
+	// The max limit is set to 5.
+	store, err := setUpStore(queueDir, 5)
+	if err != nil {
+		t.Fatal("Failed to create a queue store ", err)
+	}
+	for i := 0; i < 5; i++ {
+		if err := store.Put(testEvent); err != nil {
+			t.Fatal("Failed to put to queue store ", err)
+		}
+	}
+	// Should not allow 6th Put.
+	if err := store.Put(testEvent); err == nil {
+		t.Fatalf("Expected to fail with %s, but passes", ErrLimitExceeded)
+	}
+}
@@ -0,0 +1,37 @@
+/*
+ * Minio Cloud Storage, (C) 2019 Minio, Inc.
+ *
+ * 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://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 target
+
+import (
+	"errors"
+	"github.com/minio/minio/pkg/event"
+)
+
+// ErrLimitExceeded error is sent when the maximum limit is reached.
+var ErrLimitExceeded = errors.New("[Store] The maximum limit reached")
+
+// ErrNoSuchKey error is sent in Get when the key is not found.
+var ErrNoSuchKey = errors.New("[Store] No such key found")
+
+// Store - To persist the events.
+type Store interface {
+	Put(event event.Event) error
+	Get(key string) (event.Event, error)
+	ListAll() []string
+	Del(key string)
+	Open() error
+}
@@ -138,6 +138,12 @@ func NewReader(readCloser io.ReadCloser, args *ReaderArgs) (*Reader, error) {
 	csvReader.Comma = []rune(args.FieldDelimiter)[0]
 	csvReader.Comment = []rune(args.CommentCharacter)[0]
 	csvReader.FieldsPerRecord = -1
+	// If LazyQuotes is true, a quote may appear in an unquoted field and a
+	// non-doubled quote may appear in a quoted field.
+	csvReader.LazyQuotes = true
+	// If TrimLeadingSpace is true, leading white space in a field is ignored.
+	// This is done even if the field delimiter, Comma, is white space.
+	csvReader.TrimLeadingSpace = true

 	r := &Reader{
 		args:       args,
@@ -71,7 +71,7 @@ func (r *Reader) Close() error {

 // NewReader - creates new JSON reader using readCloser.
 func NewReader(readCloser io.ReadCloser, args *ReaderArgs) *Reader {
-	d := jstream.NewDecoder(readCloser, 0)
+	d := jstream.NewDecoder(readCloser, 0).ObjectAsKVS()
 	return &Reader{
 		args:       args,
 		decoder:    d,
@@ -42,8 +42,8 @@ type KV struct {
 // KVS - represents key values in an JSON object
 type KVS []KV

-// MarshalJSON - implements converting slice of KVS into a JSON object
-// with multiple keys and values.
+// MarshalJSON - implements converting a KVS datastructure into a JSON
+// object with multiple keys and values.
 func (kvs KVS) MarshalJSON() ([]byte, error) {
 	b := new(bytes.Buffer)
 	b.Write([]byte("{"))
@@ -69,6 +69,7 @@ type Decoder struct {
 	emitDepth     int
 	emitKV        bool
 	emitRecursive bool
+	objectAsKVS   bool

 	depth   int
 	scratch *scratch
@@ -97,6 +98,15 @@ func NewDecoder(r io.Reader, emitDepth int) *Decoder {
 	return d
 }

+// ObjectAsKVS - by default JSON returns map[string]interface{} this
+// is usually fine in most cases, but when you need to preserve the
+// input order its not a right data structure. To preserve input
+// order please use this option.
+func (d *Decoder) ObjectAsKVS() *Decoder {
+	d.objectAsKVS = true
+	return d
+}
+
 // EmitKV enables emitting a jstream.KV struct when the items(s) parsed
 // at configured emit depth are within a JSON object. By default, only
 // the object values are emitted.
@@ -217,7 +227,13 @@ func (d *Decoder) any() (interface{}, ValueType, error) {
 		i, err := d.array()
 		return i, Array, err
 	case '{':
-		i, err := d.object()
+		var i interface{}
+		var err error
+		if d.objectAsKVS {
+			i, err = d.objectOrdered()
+		} else {
+			i, err = d.object()
+		}
 		return i, Object, err
 	default:
 		return nil, Unknown, d.mkError(ErrSyntax, "looking for beginning of value")
@@ -442,7 +458,92 @@ out:
 }

 // object accept valid JSON array value
-func (d *Decoder) object() (KVS, error) {
+func (d *Decoder) object() (map[string]interface{}, error) {
+	d.depth++
+
+	var (
+		c   byte
+		k   string
+		v   interface{}
+		t   ValueType
+		err error
+		obj map[string]interface{}
+	)
+
+	// skip allocating map if it will not be emitted
+	if d.depth > d.emitDepth {
+		obj = make(map[string]interface{})
+	}
+
+	// if the object has no keys
+	if c = d.skipSpaces(); c == '}' {
+		goto out
+	}
+
+scan:
+	for {
+		offset := d.pos - 1
+
+		// read string key
+		if c != '"' {
+			err = d.mkError(ErrSyntax, "looking for beginning of object key string")
+			break
+		}
+		if k, err = d.string(); err != nil {
+			break
+		}
+
+		// read colon before value
+		if c = d.skipSpaces(); c != ':' {
+			err = d.mkError(ErrSyntax, "after object key")
+			break
+		}
+
+		// read value
+		d.skipSpaces()
+		if d.emitKV {
+			if v, t, err = d.any(); err != nil {
+				break
+			}
+			if d.willEmit() {
+				d.metaCh <- &MetaValue{
+					Offset:    int(offset),
+					Length:    int(d.pos - offset),
+					Depth:     d.depth,
+					Value:     KV{k, v},
+					ValueType: t,
+				}
+			}
+		} else {
+			if v, err = d.emitAny(); err != nil {
+				break
+			}
+		}
+
+		if obj != nil {
+			obj[k] = v
+		}
+
+		// next token must be ',' or '}'
+		switch c = d.skipSpaces(); c {
+		case '}':
+			goto out
+		case ',':
+			c = d.skipSpaces()
+			goto scan
+		default:
+			err = d.mkError(ErrSyntax, "after object key:value pair")
+			goto out
+		}
+	}
+
+out:
+	d.depth--
+	return obj, err
+}
+
+// object (ordered) accept valid JSON array value
+func (d *Decoder) objectOrdered() (KVS, error) {
 	d.depth++

 	var (
@@ -517,6 +618,7 @@ scan:
 			goto scan
 		default:
 			err = d.mkError(ErrSyntax, "after object key:value pair")
+			goto out
 		}
 	}

@@ -116,10 +116,10 @@
 			"revisionTime": "2017-09-25T03:23:15Z"
 		},
 		{
-			"checksumSHA1": "mSo9Sti7F6+laUs4NLx/p23rHD0=",
+			"checksumSHA1": "YPGDETCyuMkceQvAXafFnj6ndkY=",
 			"path": "github.com/bcicen/jstream",
-			"revision": "f306cd3e1fa602a1b513114521a0d6a5a9d5c919",
-			"revisionTime": "2019-02-06T02:23:53Z"
+			"revision": "16c1f8af81c2a9967b30d365a29472126274f998",
+			"revisionTime": "2019-02-20T04:59:26Z"
 		},
 		{
 			"checksumSHA1": "0rido7hYHQtfq3UJzVT5LClLAWc=",
Author	SHA1	Message	Date
Praveen raj Mani	78d116c487	Event persistence for MQTT (#7268 ) - The events will be persisted in queueStore if `queueDir` is set. - Else, if queueDir is not set events persist in memory. The events are replayed back when the mqtt broker is back online.	2019-02-25 18:01:13 -08:00
dcharbonnier	2fc341394d	fix json syntax (#7285 )	2019-02-25 13:02:59 +05:30
Anis Elleuch	6584c7ea2b	s3: Encode StartAfter when encoding type is passed (#7281 ) In ObjectList V2, StartAfter needs to be encoded when encoding-type is specified.	2019-02-24 18:50:28 -08:00
Harshavardhana	2520e535a0	Allow lazyQuotes for certain types of CSV (#7278 ) Set lazyQuotes to true, to allow a quote to appear in an unquote field and a non-doubled quote may appear in a quoted field.	2019-02-24 06:51:02 -08:00
Anis Elleuch	5efbe8a1b3	s3: Add support of encodingType parameter (#7265 ) This commit honors encoding-type parameter in object listing, parts listing and multipart uploads listing.	2019-02-24 11:44:24 +05:30
Nitish Tiwari	dab314900d	Fix healthcheck script to wait for netstat command output (#7275 ) Fixes #7272	2019-02-23 09:32:53 -08:00
Harshavardhana	7923b83953	Support multiple-domains in MINIO_DOMAIN (#7274 ) Fixes #7173	2019-02-23 08:48:01 +05:30
Aditya Manthramurthy	80a351633f	Update vendorized bcicen/jstream (#7257 ) - Includes an error handling fix that is waiting to be merged upstream - Uses order-preserving (un)marshalling for JSON objects.	2019-02-20 23:59:23 -08:00
Harshavardhana	bedcb7442a	Write xml.Header first instead of spaces to handle XML parsers (#7253 ) Clients like AWS SDK Java and AWS cli XML parsers are unable to handle on `\r\n` characters to avoid these errors send XML header first and write white space characters instead. Also handle cases to avoid double WriteHeader calls	2019-02-21 11:50:15 +05:30
Harshavardhana	2232b0b55f	Fix mac build failure for healthcheck binary (#7263 )	2019-02-21 08:11:16 +05:30
Harshavardhana	91576d416d	Fix GetLocalPeer usage in perf handlers (#7249 ) GetLocalPeer usage should be fixed and used only once per call for not all local endpoints.	2019-02-20 16:04:55 -08:00
Minio Trusted	3aabe45fd9	Update yaml files to latest version RELEASE.2019-02-20T22-44-29Z	2019-02-20 22:49:40 +00:00