Implement Async Create Request/Response with Oplock implementation.

src/it/groovy/com/hierynomus/smbj/SMB2FileIntegrationTest.groovy

@@ -18,22 +18,30 @@ package com.hierynomus.smbj
 import com.hierynomus.msdtyp.AccessMask
 import com.hierynomus.mserref.NtStatus
 import com.hierynomus.mssmb2.SMB2CreateDisposition
+import com.hierynomus.mssmb2.SMB2OplockLevel
 import com.hierynomus.mssmb2.SMB2ShareAccess
 import com.hierynomus.mssmb2.SMBApiException
-import com.hierynomus.smb.SMBPacket
 import com.hierynomus.smbj.auth.AuthenticationContext
 import com.hierynomus.smbj.connection.Connection
+import com.hierynomus.smbj.event.AsyncCreateResponseNotification
+import com.hierynomus.smbj.event.OplockBreakNotification
+import com.hierynomus.smbj.event.handler.AbstractNotificationHandler
+import com.hierynomus.smbj.event.handler.MessageIdCallback
 import com.hierynomus.smbj.io.ArrayByteChunkProvider
 import com.hierynomus.smbj.session.Session
+import com.hierynomus.smbj.share.DiskEntry
 import com.hierynomus.smbj.share.DiskShare
 import com.hierynomus.smbj.transport.tcp.async.AsyncDirectTcpTransportFactory
 import spock.lang.Specification
 import spock.lang.Unroll
 
 import java.nio.charset.StandardCharsets
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicBoolean
 
 import static com.hierynomus.mssmb2.SMB2CreateDisposition.FILE_CREATE
 import static com.hierynomus.mssmb2.SMB2CreateDisposition.FILE_OPEN
+import static com.hierynomus.mssmb2.SMB2CreateDisposition.FILE_OPEN_IF
 
 class SMB2FileIntegrationTest extends Specification {
 
@@ -185,4 +193,170 @@ class SMB2FileIntegrationTest extends Specification {
     cleanup:
     share.rm("bigfile")
   }
+
+  def "should able to async create"() {
+    given:
+    def path = "createAsync.txt"
+    // In actual implementation, the path is not available for createResponse complete. Map is required.
+    def messageIdPathMap = new ConcurrentHashMap<Long, String>()
+    // Should call async listener, just calling dummy in test case
+    def testSucceed = new AtomicBoolean(false)
+    share.setNotificationHandler( new AbstractNotificationHandler() {
+
+      @Override
+      void handleAsyncCreateResponseNotification(
+        AsyncCreateResponseNotification asyncCreateResponseNotification) {
+        def createResponseFuture = asyncCreateResponseNotification.future
+        def createResponse
+        try {
+          createResponse = createResponseFuture.get()
+        } catch (Throwable t) {
+          throw new IllegalStateException("Unable to get create response", t)
+        }
+        def getPath = messageIdPathMap.remove(createResponse.header.messageId)
+        if(getPath == null) {
+          System.out.println("Could not find path in map. Should not related to async create, ignored.")
+          return
+        }
+
+        if(createResponse.header.status != NtStatus.STATUS_SUCCESS) {
+          throw new IllegalStateException("Async create failed with status " + createResponse.header.status.value)
+        }
+
+        def diskEntry = share.getDiskEntry(getPath, new DiskShare.SMB2CreateResponseContext(createResponse, share))
+
+        if(diskEntry != null) {
+          // Should call async listener, just calling dummy in test case
+          testSucceed.compareAndSet(false, true)
+        }
+      }
+
+    })
+
+    when:
+    share.openAsync(path, null, null, EnumSet.of(AccessMask.GENERIC_READ, AccessMask.GENERIC_WRITE), null, SMB2ShareAccess.ALL, FILE_CREATE, null, new MessageIdCallback() {
+
+      @Override
+      void callback(long messageId) {
+        messageIdPathMap.put(messageId, path)
+      }
+    })
+
+    then:
+    // 1 second should be enough for the whole process complete in docker
+    Thread.sleep(1000L)
+
+    expect:
+    testSucceed.get() == true
+
+    cleanup:
+    share.rm(path)
+    messageIdPathMap.clear()
+
+  }
+
+  def "should able to receive oplock break notification and response acknowledgement then receive acknowledgement response"() {
+    given:
+    def path = "createAsyncOplock.txt"
+    // In actual implementation, the path is not available for createResponse complete. Map is required.
+    def messageIdPathMap = new ConcurrentHashMap<Long, String>()
+    // Should call async listener, just using hashmap as dummy in test case
+    def messageIdDiskEntryMap = new ConcurrentHashMap<Long, DiskEntry>()
+    def fileIdDiskEntryMap = new ConcurrentHashMap<String, DiskEntry>()
+    def succeedBreakToLevel2 = new AtomicBoolean(false)
+    def oplockBreakAcknowledgmentResponseSucceed = new AtomicBoolean(false)
+    share.setNotificationHandler( new AbstractNotificationHandler() {
+
+      @Override
+      void handleAsyncCreateResponseNotification(
+        AsyncCreateResponseNotification asyncCreateResponseNotification) {
+        def createResponseFuture = asyncCreateResponseNotification.future
+        def createResponse
+        try {
+          createResponse = createResponseFuture.get()
+        } catch (Throwable t) {
+          throw new IllegalStateException("Unable to get create response", t)
+        }
+        def getPath = messageIdPathMap.remove(createResponse.header.messageId)
+        if(getPath == null) {
+          System.out.println("Could not find path in map. Should not related to async create, ignored.")
+          return
+        }
+
+        if(createResponse.header.status != NtStatus.STATUS_SUCCESS) {
+          throw new IllegalStateException("Async create failed with status " + createResponse.header.status.value)
+        }
+
+        def diskEntry = share.getDiskEntry(getPath, new DiskShare.SMB2CreateResponseContext(createResponse, share))
+
+        if(diskEntry != null) {
+          // Should call async listener, just calling dummy in test case
+          messageIdDiskEntryMap.put(createResponse.header.messageId, diskEntry)
+          fileIdDiskEntryMap.put(diskEntry.fileId.toHexString(), diskEntry)
+        }
+      }
+
+      @Override
+      void handleOplockBreakNotification(OplockBreakNotification oplockBreakNotification) {
+        def oplockBreakLevel = oplockBreakNotification.oplockLevel
+        def getDiskEntry = fileIdDiskEntryMap.get(oplockBreakNotification.fileId.toHexString())
+        if(getDiskEntry == null) {
+          throw new IllegalStateException("Unable to get corresponding diskEntry!")
+        }
+        // Assume we already notify client and had succeed handled client cache to break
+        if(oplockBreakLevel) {
+          // In this test case, this code should only run exactly once.
+          succeedBreakToLevel2.compareAndSet(false, true)
+        }
+        // Should return to client for handling the client cache, dummy in test case
+        def oplockBreakAcknowledgmentResponse = getDiskEntry.acknowledgeOplockBreak(oplockBreakLevel)
+        if(oplockBreakAcknowledgmentResponse.header.status == NtStatus.STATUS_SUCCESS) {
+          // In this test case, this code should only run exactly once.
+          oplockBreakAcknowledgmentResponseSucceed.compareAndSet(false, true)
+        }
+      }
+    })
+
+    when:
+    def firstCreateMessageId = 0L
+    share.openAsync(path, SMB2OplockLevel.SMB2_OPLOCK_LEVEL_EXCLUSIVE, null, EnumSet.of(AccessMask.GENERIC_READ, AccessMask.GENERIC_WRITE), null, SMB2ShareAccess.ALL, FILE_OPEN_IF, null, new MessageIdCallback() {
+
+      @Override
+      void callback(long messageId) {
+        messageIdPathMap.put(messageId, path)
+        firstCreateMessageId = messageId
+      }
+    })
+
+    then:
+    // 1 second should be enough for the whole process complete in docker
+    Thread.sleep(1000L)
+    def firstCreateDiskEntry = messageIdDiskEntryMap.remove(firstCreateMessageId)
+    // another create to the same file with SMB2_OPLOCK_LEVEL_EXCLUSIVE to trigger oplock break notification in Server.
+    def secondCreateMessageId = 0L
+    share.openAsync(path, SMB2OplockLevel.SMB2_OPLOCK_LEVEL_EXCLUSIVE, null, EnumSet.of(AccessMask.GENERIC_READ, AccessMask.GENERIC_WRITE), null, SMB2ShareAccess.ALL, FILE_OPEN_IF, null, new MessageIdCallback() {
+
+      @Override
+      void callback(long messageId) {
+        messageIdPathMap.put(messageId, path)
+        secondCreateMessageId = messageId
+      }
+    })
+    // 1 second should be enough for the whole process complete in docker
+    Thread.sleep(1000L)
+    def secondCreateDiskEntry = messageIdDiskEntryMap.remove(secondCreateMessageId)
+
+    expect:
+    firstCreateDiskEntry != null
+    secondCreateDiskEntry != null
+    succeedBreakToLevel2.get() == true
+    oplockBreakAcknowledgmentResponseSucceed.get() == true
+
+    cleanup:
+    share.rm(path)
+    messageIdPathMap.clear()
+    messageIdDiskEntryMap.clear()
+    fileIdDiskEntryMap.clear()
+
+  }
 }

src/main/java/com/hierynomus/mserref/NtStatus.java

@@ -65,6 +65,7 @@ public enum NtStatus implements EnumWithValue<NtStatus> {
     STATUS_NOT_SAME_DEVICE(0xC00000D4L),
     STATUS_FILE_RENAMED(0xC00000D5L),
     STATUS_OPLOCK_NOT_GRANTED(0xC00000E2L),
+    STATUS_INVALID_OPLOCK_PROTOCOL(0xC00000E3L),
     STATUS_INTERNAL_ERROR(0xC00000E5L),
     STATUS_UNEXPECTED_IO_ERROR(0xC00000E9L),
     STATUS_DIRECTORY_NOT_EMPTY(0xC0000101L),

src/main/java/com/hierynomus/mssmb2/SMB2FileId.java

@@ -16,9 +16,12 @@
 package com.hierynomus.mssmb2;
 
 import com.hierynomus.protocol.commons.ByteArrayUtils;
+import com.hierynomus.protocol.commons.Objects;
 import com.hierynomus.protocol.commons.buffer.Buffer;
 import com.hierynomus.smb.SMBBuffer;
 
+import java.util.Arrays;
+
 /**
  * [MS-SMB2].pdf 2.2.14.1 SMB2_FILEID
  */
@@ -53,4 +56,24 @@ public String toString() {
             "persistentHandle=" + ByteArrayUtils.printHex(persistentHandle) +
             '}';
     }
+
+    public String toHexString() {
+        return ByteArrayUtils.toHex(persistentHandle) + ByteArrayUtils.toHex(volatileHandle);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        SMB2FileId smb2FileId = (SMB2FileId) o;
+        return Objects.equals(persistentHandle, smb2FileId.persistentHandle) &&
+               Objects.equals(volatileHandle, smb2FileId.volatileHandle);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Arrays.hashCode(persistentHandle);
+        result = 31 * result + Arrays.hashCode(volatileHandle);
+        return result;
+    }
 }

src/main/java/com/hierynomus/mssmb2/SMB2OplockBreakLevel.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright (C)2016 - SMBJ Contributors
+ *
+ * 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 com.hierynomus.mssmb2;
+
+import com.hierynomus.protocol.commons.EnumWithValue;
+
+/**
+ * [MS-SMB2].pdf 2.2.23 SMB2 OPLOCK_BREAK Notification - OplockLevel
+ */
+public enum SMB2OplockBreakLevel implements EnumWithValue<SMB2OplockBreakLevel> {
+    SMB2_OPLOCK_LEVEL_NONE(0x00L),
+    SMB2_OPLOCK_LEVEL_II(0x01L);
+
+    private long value;
+
+    SMB2OplockBreakLevel(long value) {
+        this.value = value;
+    }
+
+    @Override
+    public long getValue() {
+        return value;
+    }
+}

src/main/java/com/hierynomus/mssmb2/SMB2OplockLevel.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C)2016 - SMBJ Contributors
+ *
+ * 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 com.hierynomus.mssmb2;
+
+import com.hierynomus.protocol.commons.EnumWithValue;
+
+/**
+ * [MS-SMB2].pdf 2.2.13 SMB2 CREATE Request - OplockLevel
+ * <p>
+ */
+public enum SMB2OplockLevel implements EnumWithValue<SMB2OplockLevel> {
+    SMB2_OPLOCK_LEVEL_NONE(0x00L),
+    SMB2_OPLOCK_LEVEL_II(0x01L),
+    SMB2_OPLOCK_LEVEL_EXCLUSIVE(0x08L),
+    SMB2_OPLOCK_LEVEL_BATCH(0x09L),
+    // TODO: implement and support using lease
+    OPLOCK_LEVEL_LEASE(0xFFL);
+
+    private long value;
+
+    SMB2OplockLevel(long value) {
+        this.value = value;
+    }
+
+    @Override
+    public long getValue() {
+        return value;
+    }
+}

src/main/java/com/hierynomus/mssmb2/messages/SMB2CreateRequest.java

@@ -39,16 +39,19 @@ public class SMB2CreateRequest extends SMB2Packet {
     private final SmbPath path;
     private final Set<AccessMask> accessMask;
     private final SMB2ImpersonationLevel impersonationLevel;
+    private final SMB2OplockLevel oplockLevel;
 
     @SuppressWarnings("PMD.ExcessiveParameterList")
     public SMB2CreateRequest(SMB2Dialect smbDialect,
                              long sessionId, long treeId,
+                             SMB2OplockLevel oplockLevel,
                              SMB2ImpersonationLevel impersonationLevel,
                              Set<AccessMask> accessMask,
                              Set<FileAttributes> fileAttributes,
                              Set<SMB2ShareAccess> shareAccess, SMB2CreateDisposition createDisposition,
                              Set<SMB2CreateOptions> createOptions, SmbPath path) {
         super(57, smbDialect, SMB2MessageCommandCode.SMB2_CREATE, sessionId, treeId);
+        this.oplockLevel = ensureNotNull(oplockLevel, SMB2OplockLevel.SMB2_OPLOCK_LEVEL_NONE);
         this.impersonationLevel = ensureNotNull(impersonationLevel, SMB2ImpersonationLevel.Identification);
         this.accessMask = accessMask;
         this.fileAttributes = ensureNotNull(fileAttributes, FileAttributes.class);
@@ -62,7 +65,7 @@ public SMB2CreateRequest(SMB2Dialect smbDialect,
     protected void writeTo(SMBBuffer buffer) {
         buffer.putUInt16(structureSize); // StructureSize (2 bytes)
         buffer.putByte((byte) 0); // SecurityFlags (1 byte) - Reserved
-        buffer.putByte((byte) 0);  // RequestedOpLockLevel (1 byte) - None
+        buffer.putByte((byte)oplockLevel.getValue());  // RequestedOpLockLevel (1 byte)
         buffer.putUInt32(impersonationLevel.getValue()); // ImpersonationLevel (4 bytes) - Identification
         buffer.putReserved(8); // SmbCreateFlags (8 bytes)
         buffer.putReserved(8); // Reserved (8 bytes)
@@ -95,4 +98,8 @@ protected void writeTo(SMBBuffer buffer) {
 
         buffer.putRawBytes(nameBytes);
     }
+
+    public SmbPath getPath() {
+        return path;
+    }
 }