Making a semantic change to queue client, subscription client setPrefetchCount so prefetch count can be set even without register a handler. (#107)

* Fixing javadoc generation errors

* Adding dependency to use SLF4J logging instead of java util logging.

* More SLF4J migration

* Changed all java util logging statements to slf4j.

* Checking in the POM that goes with the release to maven reposisory.

* Removed unused imports

* Instrumented all code with SLF4J logging.

* Fixing some log statements

* Moving request-response link creation to message factory so we don't create multiple links to the same entity.

* Updaing third party notice to include SLF4J license.

* Updating distributed POM

* Renaming setContent method to setBody.

* Fixing a typo in log statement.

* Handling the special case of 0 timeout for receives. If there are no prefetched messages, receive call returns immediately with already prefetched messages.

* Correcting a log statement

* Added javadoc comments for rules package.

* Adding javadoc comments. Work in progress.

* More javadoc comments added.

* Fixing a minor bug in sender and receiver creation. Closes the messaging factory if sender or receiver creation fails.

* Fixing the issue of not renewing CBS tokens. Also a concurrency fix.

* Fixing a bug in session receiver to not repeatedly attempt accepting the same session on lock lost.

* Added some java docs.. And changed default prefetch count, based on receive mode.

* Fixing a thread unending wait bug in request-response link.

* Fixing AuthorizationFailed exceptions that pop when the conncetion is recreated.

* Minor tweaks

* Another minor fix

* More minor bug fixes and tweaks

* Changing version to 1.0.0-RC-1

* Fixing some concurrency issues, performance improvements and refactoring.

* Minor fix in receiver

* Changing return messages loop interval to 100 milliseconds

* Changing zero timeout approximation to 200 milliseconds

* Changing zero timeout approximation to 200 milliseconds. Also improving tracing to better identify stress issues.

* Adding more info to a trace

* Changing default prefetch count to 0 in peeklock mode as well.

* Fixing a bug in setPrefetchCount

* Chagning version number to 1.0.0

* Making a semantic change to queue client, subscription client setPrefetchCount so prefetch count can be set even without register a handler.

Author: yvgopal

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/MessageAndSessionPump.java

@@ -32,6 +32,7 @@ class MessageAndSessionPump extends InitializableEntity implements IMessageAndSe
     private static final Duration MINIMUM_MESSAGE_LOCK_VALIDITY = Duration.ofSeconds(4);
     private static final Duration MAXIMUM_RENEW_LOCK_BUFFER = Duration.ofSeconds(10);
     private static final Duration SLEEP_DURATION_ON_ACCEPT_SESSION_EXCEPTION = Duration.ofMinutes(1);
+    private static final int UNSET_PREFETCH_COUNT = -1; // Means prefetch count not set
 
     private final ConcurrentHashMap<String, IMessageSession> openSessions;
     private final MessagingFactory factory;
@@ -44,13 +45,15 @@ class MessageAndSessionPump extends InitializableEntity implements IMessageAndSe
     private ISessionHandler sessionHandler;
     private MessageHandlerOptions messageHandlerOptions;
     private SessionHandlerOptions sessionHandlerOptions;
+    private int prefetchCount;
 
     public MessageAndSessionPump(MessagingFactory factory, String entityPath, ReceiveMode receiveMode) {
         super(StringUtil.getShortRandomString(), null);
         this.factory = factory;
         this.entityPath = entityPath;
         this.receiveMode = receiveMode;
         this.openSessions = new ConcurrentHashMap<>();
+        this.prefetchCount = UNSET_PREFETCH_COUNT;
     }
 
     @Override
@@ -67,6 +70,10 @@ public void registerMessageHandler(IMessageHandler handler, MessageHandlerOption
 
         this.innerReceiver = ClientFactory.createMessageReceiverFromEntityPath(this.factory, this.entityPath, this.receiveMode);
         TRACE_LOGGER.info("Created MessageReceiver to entity '{}'", this.entityPath);
+        if(this.prefetchCount != UNSET_PREFETCH_COUNT)
+        {
+            this.innerReceiver.setPrefetchCount(this.prefetchCount);
+        }
         for (int i = 0; i < handlerOptions.getMaxConcurrentCalls(); i++) {
             this.receiveAndPumpMessage();
         }
@@ -221,6 +228,14 @@ private void acceptSessionsAndPumpMessage() {
                 } else {
                     // Received a session.. Now pump messages..
                     TRACE_LOGGER.debug("Accepted a session '{}' from entity '{}'", session.getSessionId(), this.entityPath);
+                    if(this.prefetchCount != UNSET_PREFETCH_COUNT)
+                    {
+                        try {
+                            session.setPrefetchCount(this.prefetchCount);
+                        } catch (ServiceBusException e) {
+                            // Should not happen as long as reactor is running. So ignoring
+                        }
+                    }
                     this.openSessions.put(session.getSessionId(), session);
                     SessionRenewLockLoop sessionRenewLockLoop = new SessionRenewLockLoop(session, this);
                     sessionRenewLockLoop.startLoop();
@@ -713,13 +728,34 @@ private void notifyExceptionToMessageHandler(Throwable ex, ExceptionPhase phase)
 
     @Override
     public int getPrefetchCount() {
-        this.checkInnerReceiveCreated();
-        return this.innerReceiver.getPrefetchCount();
+        return this.prefetchCount;
     }
 
     @Override
     public void setPrefetchCount(int prefetchCount) throws ServiceBusException {
-        this.checkInnerReceiveCreated();
-        this.innerReceiver.setPrefetchCount(prefetchCount);
+        if(prefetchCount < 0)
+        {
+            throw new IllegalArgumentException("Prefetch count cannot be negative.");
+        }
+        
+        this.prefetchCount = prefetchCount;
+        if(this.innerReceiver != null)
+        {
+            this.innerReceiver.setPrefetchCount(prefetchCount);
+        }
+
+        // For accepted session receivers also
+        IMessageSession[] currentAcceptedSessions = this.openSessions.values().toArray(new IMessageSession[0]);
+        for(IMessageSession session : currentAcceptedSessions)
+        {
+            try
+            {
+                session.setPrefetchCount(prefetchCount);
+            }
+            catch(IllegalStateException ise)
+            {
+                // Session might have been closed.. Ignore the exception as this is a best effort setter on already accepted sessions
+            }
+        }
     }
 }

azure-servicebus/src/main/java/com/microsoft/azure/servicebus/primitives/CoreMessageReceiver.java

@@ -471,7 +471,7 @@ public void setPrefetchCount(final int value) throws ServiceBusException
 		if(deltaPrefetchCount > 0)
 		{
 		    try
-	        {	            
+	        {
 	            this.underlyingFactory.scheduleOnReactorThread(new DispatchHandler()
 	            {
 	                @Override