Add support of listen bucket notification (#727)

Also, change header/query map type to follow HTTP spec

One header in HTTP can hold several values.
   ```
    Header: Val1
    Header: Val2
   ```

The same thing goes for query:
   `GET /path/?key=val1&key=val2`

This is needed when implementing listen bucket notification, because
this API accept multiple events under the same query name.

Author: vadmeste

api/src/main/java/io/minio/BucketEventListener.java

@@ -0,0 +1,26 @@
+/*
+ * Minio Java SDK for Amazon S3 Compatible Cloud Storage,
+ * (C) 2018 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 io.minio;
+
+import io.minio.notification.NotificationInfo;
+
+public interface BucketEventListener {
+
+  void updateEvent(NotificationInfo info);
+}
+

api/src/main/java/io/minio/MinioClient.java

@@ -17,6 +17,9 @@
 
 package io.minio;
 
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Multimaps;
 import com.google.common.io.ByteStreams;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -57,6 +60,8 @@
 import io.minio.messages.NotificationConfiguration;
 import io.minio.org.apache.commons.validator.routines.InetAddressValidator;
 
+import io.minio.notification.NotificationInfo;
+
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
@@ -69,6 +74,8 @@
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmlpull.v1.XmlPullParserFactory;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -815,8 +822,8 @@ public boolean verify(String hostname, SSLSession session) {
    * @param length         Length of HTTP request body.
    */
   private Request createRequest(Method method, String bucketName, String objectName,
-                                String region, Map<String,String> headerMap,
-                                Map<String,String> queryParamMap, final String contentType,
+                                String region, Multimap<String,String> headerMap,
+                                Multimap<String,String> queryParamMap, final String contentType,
                                 Object body, int length)
     throws InvalidBucketNameException, NoSuchAlgorithmException, InvalidKeyException, InsufficientDataException,
            IOException, InternalException {
@@ -870,7 +877,7 @@ private Request createRequest(Method method, String bucketName, String objectNam
     }
 
     if (queryParamMap != null) {
-      for (Map.Entry<String,String> entry : queryParamMap.entrySet()) {
+      for (Map.Entry<String,String> entry : queryParamMap.entries()) {
         urlBuilder.addEncodedQueryParameter(S3Escaper.encode(entry.getKey()), S3Escaper.encode(entry.getValue()));
       }
     }
@@ -880,7 +887,7 @@ private Request createRequest(Method method, String bucketName, String objectNam
     Request.Builder requestBuilder = new Request.Builder();
     requestBuilder.url(url);
     if (headerMap != null) {
-      for (Map.Entry<String,String> entry : headerMap.entrySet()) {
+      for (Map.Entry<String,String> entry : headerMap.entries()) {
         requestBuilder.header(entry.getKey(), entry.getValue());
       }
     }
@@ -998,9 +1005,29 @@ private HttpResponse execute(Method method, String region, String bucketName, St
     throws InvalidBucketNameException, NoSuchAlgorithmException, InsufficientDataException, IOException,
            InvalidKeyException, NoResponseException, XmlPullParserException, ErrorResponseException,
            InternalException {
-    String contentType = null;
+
+    Multimap<String, String> queryParamMultiMap = null;
+    if (queryParamMap != null) {
+      queryParamMultiMap = Multimaps.forMap(queryParamMap);
+    }
+
+    Multimap<String, String> headerMultiMap = null;
     if (headerMap != null) {
-      contentType = headerMap.get("Content-Type");
+      headerMultiMap = Multimaps.forMap(headerMap);
+    }
+
+    return executeReq(method, region, bucketName, objectName, headerMultiMap, queryParamMultiMap, body, length);
+  }
+
+  private HttpResponse executeReq(Method method, String region, String bucketName, String objectName,
+                               Multimap<String,String> headerMap, Multimap<String,String> queryParamMap,
+                               Object body, int length)
+    throws InvalidBucketNameException, NoSuchAlgorithmException, InsufficientDataException, IOException,
+           InvalidKeyException, NoResponseException, XmlPullParserException, ErrorResponseException,
+           InternalException {
+    String contentType = null;
+    if (headerMap != null && headerMap.get("Content-Type") != null) {
+      contentType = String.join(" ", headerMap.get("Content-Type"));
     }
     if (body != null && !(body instanceof InputStream || body instanceof RandomAccessFile || body instanceof byte[])) {
       byte[] bytes = body.toString().getBytes(StandardCharsets.UTF_8);
@@ -2053,8 +2080,16 @@ public String getPresignedObjectUrl(Method method, String bucketName, String obj
       body = new byte[0];
     }
 
+    Multimap<String, String> queryParamMap = null;
+    if (reqParams != null) {
+      queryParamMap = HashMultimap.create();
+      for (Map.Entry<String, String> m: reqParams.entrySet()) {
+        queryParamMap.put(m.getKey(), m.getValue());
+      }
+    }
+
     String region = getRegion(bucketName);
-    Request request = createRequest(method, bucketName, objectName, region, null, reqParams, null, body, 0);
+    Request request = createRequest(method, bucketName, objectName, region, null, queryParamMap, null, body, 0);
     HttpUrl url = Signer.presignV4(request, region, accessKey, secretKey, expires);
     return url.toString();
   }
@@ -4303,6 +4338,63 @@ public void removeIncompleteUpload(String bucketName, String objectName)
     }
   }
 
+  /**
+   * Listen to bucket notifications.
+   *
+   * @param bucketName Bucket name.
+   * @param prefix Prefix of concerned objects events.
+   * @param suffix Suffix of concerned objects events.
+   * @param events List of events to watch.
+   * @param eventCallback Event handler.
+   *
+   */
+
+  public void listenBucketNotification(String bucketName, String prefix, String suffix, String[] events,
+      BucketEventListener eventCallback)
+    throws InvalidBucketNameException, NoSuchAlgorithmException, InsufficientDataException, IOException,
+                    InvalidKeyException, NoResponseException, XmlPullParserException, ErrorResponseException,
+                    InternalException  {
+
+    Multimap<String,String> queryParamMap = HashMultimap.create();
+    queryParamMap.put("prefix", prefix);
+    queryParamMap.put("suffix", suffix);
+    for (String event: events) {
+      queryParamMap.put("events", event);
+    }
+
+    String bodyContent = "";
+    Scanner scanner = null;
+    HttpResponse response = null;
+    ObjectMapper mapper = new ObjectMapper();
+
+    try {
+      response = executeReq(Method.GET, getRegion(bucketName),
+          bucketName, "", null, queryParamMap, null, 0);
+      scanner = new Scanner(response.body().charStream());
+      scanner.useDelimiter("\n");
+      while (scanner.hasNext()) {
+        bodyContent = scanner.next().trim();
+        if (bodyContent.equals("")) {
+          continue;
+        }
+        NotificationInfo ni = mapper.readValue(bodyContent, NotificationInfo.class);
+        eventCallback.updateEvent(ni);
+      }
+    } catch (RuntimeException e) {
+      throw e;
+    } catch (Exception e) {
+      throw e;
+    } finally {
+      if (response != null) {
+        response.body().close();
+      }
+      if (scanner != null) {
+        scanner.close();
+      }
+    }
+  }
+
+
 
   /**
    * Skips data of up to given length in given input stream.
@@ -4449,3 +4541,4 @@ public void traceOff() throws IOException {
     this.traceStream = null;
   }
 }
+

api/src/main/java/io/minio/Signer.java

@@ -25,12 +25,15 @@
 import java.util.Set;
 import java.util.TreeMap;
 
+
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
 
 import org.joda.time.DateTime;
 
 import com.google.common.base.Joiner;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.MultimapBuilder;
 import com.google.common.io.BaseEncoding;
 import okhttp3.Headers;
 import okhttp3.HttpUrl;
@@ -142,16 +145,17 @@ private void setCanonicalHeaders() {
     this.signedHeaders = Joiner.on(";").join(this.canonicalHeaders.keySet());
   }
 
-
   private void setCanonicalQueryString() {
-    Map<String,String> signedQueryParams = new TreeMap<>();
-
     String encodedQuery = this.url.encodedQuery();
     if (encodedQuery == null) {
       this.canonicalQueryString = "";
       return;
     }
 
+    // Building a multimap which only order keys, ordering values is not performed
+    // until Minio server supports it.
+    Multimap<String, String> signedQueryParams = MultimapBuilder.treeKeys().arrayListValues().build();
+
     for (String queryParam : encodedQuery.split("&")) {
       String[] tokens = queryParam.split("=");
       if (tokens.length > 1) {
@@ -161,10 +165,9 @@ private void setCanonicalQueryString() {
       }
     }
 
-    this.canonicalQueryString = Joiner.on("&").withKeyValueSeparator("=").join(signedQueryParams);
+    this.canonicalQueryString = Joiner.on("&").withKeyValueSeparator("=").join(signedQueryParams.entries());
   }
 
-
   private void setCanonicalRequest() throws NoSuchAlgorithmException {
     setCanonicalHeaders();
     this.url = this.request.url();

api/src/main/java/io/minio/messages/EventType.java

@@ -31,6 +31,9 @@ public enum EventType {
   OBJECT_CREATED_POST("s3:ObjectCreated:Post"),
   OBJECT_CREATED_COPY("s3:ObjectCreated:Copy"),
   OBJECT_CREATED_COMPLETE_MULTIPART_UPLOAD("s3:ObjectCreated:CompleteMultipartUpload"),
+  OBJECT_ACCESSED_GET("s3:ObjectAccessed:Get"),
+  OBJECT_ACCESSED_HEAD("s3:ObjectAccessed:Head"),
+  OBJECT_ACCESSED_ANY("s3:ObjectAccessed:*"),
   OBJECT_REMOVED_ANY("s3:ObjectRemoved:*"),
   OBJECT_REMOVED_DELETE("s3:ObjectRemoved:Delete"),
   OBJECT_REMOVED_DELETED_MARKER_CREATED("s3:ObjectRemoved:DeleteMarkerCreated"),

api/src/main/java/io/minio/notification/BucketMeta.java

@@ -0,0 +1,31 @@
+/*
+ * Minio Java SDK for Amazon S3 Compatible Cloud Storage,
+ * (C) 2018 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 io.minio.notification;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+@SuppressFBWarnings("UUF_UNUSED_PUBLIC_OR_PROTECTED_FIELD")
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class BucketMeta {
+  public String name;
+  public Identity ownerIdentity;
+  public String arn;
+}
+
+

api/src/main/java/io/minio/notification/EventMeta.java

@@ -0,0 +1,32 @@
+/*
+ * Minio Java SDK for Amazon S3 Compatible Cloud Storage,
+ * (C) 2018 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 io.minio.notification;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+@SuppressFBWarnings("UUF_UNUSED_PUBLIC_OR_PROTECTED_FIELD")
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class EventMeta {
+  public String schemaVersion;
+  public String configurationId;
+  public BucketMeta bucket;
+  public ObjectMeta object;
+}
+
+

api/src/main/java/io/minio/notification/Identity.java

@@ -0,0 +1,29 @@
+/*
+ * Minio Java SDK for Amazon S3 Compatible Cloud Storage,
+ * (C) 2018 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 io.minio.notification;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+@SuppressFBWarnings("UUF_UNUSED_PUBLIC_OR_PROTECTED_FIELD")
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class Identity {
+  public String principalId;
+}
+
+

api/src/main/java/io/minio/notification/NotificationEvent.java

@@ -0,0 +1,38 @@
+/*
+ * Minio Java SDK for Amazon S3 Compatible Cloud Storage,
+ * (C) 2018 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 io.minio.notification;
+
+import java.util.Map;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+
+@SuppressFBWarnings("UUF_UNUSED_PUBLIC_OR_PROTECTED_FIELD")
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class NotificationEvent {
+  public String eventVersion;
+  public String eventSource;
+  public String awsRegion;
+  public String eventTime;
+  public String eventName;
+  public Identity userIdentity;
+  public Map<String, String> requestParameters;
+  public Map<String, String> responseElements;
+  public EventMeta s3;
+  public SourceInfo source;
+}
+