Reworked the obeyFloodLimit-Stuff to now use lambdas.

src/main/kotlin/de/fabianonline/telegram_backup/DownloadManager.kt

@@ -61,36 +61,11 @@ enum class MessageSource(val descr: String) {
 }
 
 class DownloadManager(val client: TelegramClient, val prog: DownloadProgressInterface, val db: Database, val user_manager: UserManager, val settings: Settings, val file_base: String) {
-	internal var has_seen_flood_wait_message = false
-
 	@Throws(RpcErrorException::class, IOException::class)
 	fun downloadMessages(limit: Int?) {
-		var completed: Boolean
+		Utils.obeyFloodWait() {
-		do {
-			completed = true
-			try {
 			_downloadMessages(limit)
-			} catch (e: RpcErrorException) {
-				if (e.getCode() == 420) { // FLOOD_WAIT
-					completed = false
-					Utils.obeyFloodWaitException(e)
-				} else {
-					throw e
 		}
-			} catch (e: TimeoutException) {
-				completed = false
-				System.out.println("")
-				System.out.println("Telegram took too long to respond to our request.")
-				System.out.println("I'm going to wait a minute and then try again.")
-				try {
-					TimeUnit.MINUTES.sleep(1)
-				} catch (e2: InterruptedException) {
-				}
-
-				System.out.println("")
-			}
-
-		} while (!completed)
 	}
 
 	@Throws(RpcErrorException::class, IOException::class, TimeoutException::class)
@@ -211,31 +186,19 @@ class DownloadManager(val client: TelegramClient, val prog: DownloadProgressInte
 			logger.trace("vector.size(): {}", vector.size)
 			logger.trace("ids.size(): {}", ids.size)
 
-			var response: TLAbsMessages
+			var resp: TLAbsMessages? = null
-			var tries = 0
+			try {
-			while (true) {
+				Utils.obeyFloodWait(max_tries=5) {
-				logger.trace("Trying getMessages(), tries={}", tries)
+					if (channel == null) {
-				if (tries >= 5) {
+						resp = client.messagesGetMessages(vector)
+					} else {
+						resp = client.channelsGetMessages(channel, vector)
+					}
+				}
+			} catch (e: MaxTriesExceededException) {
 				CommandLineController.show_error("Couldn't getMessages after 5 tries. Quitting.")
 			}
-				tries++
+			val response = resp!!
-				try {
-					if (channel == null) {
-						response = client.messagesGetMessages(vector)
-					} else {
-						response = client.channelsGetMessages(channel, vector)
-					}
-					break
-				} catch (e: RpcErrorException) {
-					if (e.getCode() == 420) { // FLOOD_WAIT
-						Utils.obeyFloodWaitException(e, has_seen_flood_wait_message)
-						has_seen_flood_wait_message = true
-					} else {
-						throw e
-					}
-				}
-
-			}
 			logger.trace("response.getMessages().size(): {}", response.getMessages().size)
 			if (response.getMessages().size != vector.size) {
 				CommandLineController.show_error("Requested ${vector.size} messages, but got ${response.getMessages().size}. That is unexpected. Quitting.")
@@ -260,30 +223,10 @@ class DownloadManager(val client: TelegramClient, val prog: DownloadProgressInte
 	@Throws(RpcErrorException::class, IOException::class)
 	fun downloadMedia() {
 		download_client = client.getDownloaderClient()
-		var completed: Boolean
+		Utils.obeyFloodWait() {
-		do {
-			completed = true
-			try {
 			_downloadMedia()
-			} catch (e: RpcErrorException) {
-				if (e.getCode() == 420) { // FLOOD_WAIT
-					completed = false
-					Utils.obeyFloodWaitException(e)
-				} else {
-					throw e
 		}
 	}
-			/*catch (TimeoutException e) {
-				completed = false;
-				System.out.println("");
-				System.out.println("Telegram took too long to respond to our request.");
-				System.out.println("I'm going to wait a minute and then try again.");
-				logger.warn("TimeoutException caught", e);
-				try { TimeUnit.MINUTES.sleep(1); } catch(InterruptedException e2) {}
-				System.out.println("");
-			}*/
-		} while (!completed)
-	}
 
 	@Throws(RpcErrorException::class, IOException::class)
 	private fun _downloadMedia() {
@@ -423,31 +366,30 @@ class DownloadManager(val client: TelegramClient, val prog: DownloadProgressInte
 					logger.trace("offset: {} block_size: {} size: {}", offset, size, size)
 					val req = TLRequestUploadGetFile(loc, offset, size)
 					try {
+						Utils.obeyFloodWait() {
 							response = download_client!!.executeRpcQuery(req, dcID) as TLFile
+						}
 					} catch (e: RpcErrorException) {
-						if (e.getCode() == 420) { // FLOOD_WAIT
+						if (e.getCode() == 400) {
-							try_again = true
+							// Somehow this file is broken. No idea why. Let's skip it for now.
-							Utils.obeyFloodWaitException(e)
-							continue // response is null since we didn't actually receive any data. Skip the rest of this iteration and try again.
-						} else if (e.getCode() == 400) {
-							//Somehow this file is broken. No idea why. Let's skip it for now
 							return false
-						} else {
+						}
 						throw e
 					}
-					}
 					
-					offset += response.getBytes().getData().size
+					val resp = response!!
-					logger.trace("response: {} total size: {}", response.getBytes().getData().size, offset)
 					
-					fos.write(response.getBytes().getData())
+					offset += resp.getBytes().getData().size
+					logger.trace("response: {} total size: {}", resp.getBytes().getData().size, offset)
+
+					fos.write(resp.getBytes().getData())
 					fos.flush()
 					try {
 						TimeUnit.MILLISECONDS.sleep(Config.DELAY_AFTER_GET_FILE)
 					} catch (e: InterruptedException) {
 					}
 
-				} while (offset < size && (try_again || response!!.getBytes().getData().size > 0))
+				} while (offset < size && (try_again || resp.getBytes().getData().size > 0))
 				fos.close()
 				if (offset < size) {
 					System.out.println("Requested file $target with $size bytes, but got only $offset bytes.")

src/main/kotlin/de/fabianonline/telegram_backup/Utils.kt

@@ -20,6 +20,7 @@ import com.github.badoualy.telegram.tl.exception.RpcErrorException
 import java.io.File
 import java.util.Vector
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
 import com.google.gson.*
 import java.net.URL
 import org.apache.commons.io.IOUtils
@@ -32,6 +33,8 @@ object Utils {
 	@JvmField public val VERSION_1_NEWER = 1
 	@JvmField public val VERSION_2_NEWER = 2
 	
+	var hasSeenFloodWaitMessage = false
+
 	var anonymize = false
 
 	private val logger = LoggerFactory.getLogger(Utils::class.java) as Logger
@@ -97,29 +100,47 @@ object Utils {
 
 	}
 	
-	@Throws(RpcErrorException::class)
+	fun obeyFloodWait(max_tries: Int = -1, method: () -> Unit) {
-	@JvmOverloads internal fun obeyFloodWaitException(e: RpcErrorException?, silent: Boolean = false) {
+		var tries = 0
-		if (e == null || e.getCode() != 420) return
+		while (true) {
+			tries++
+			if (max_tries>0 && tries>max_tries) throw MaxTriesExceededException()
+			logger.trace("This is try ${tries}.")
+			try {
+				method.invoke()
+				// If we reach this, the method has returned successfully -> we are done
+				return
+			} catch(e: RpcErrorException) {
+				// If we got something else than a FLOOD_WAIT error, we just rethrow it
+				if (e.getCode() != 420) throw e
 				
-		val delay: Long = e.getTagInteger()!!.toLong()
+				val delay = e.getTagInteger()!!.toLong()
-		if (!silent) {
+				
-			System.out.println("")
+				if (!hasSeenFloodWaitMessage) {
-			System.out.println(
+					println(
+						"\n" +
 						"Telegram complained about us (okay, me) making too many requests in too short time by\n" +
-					"sending us \"" + e.getTag() + "\" as an error. So we now have to wait a bit. Telegram\n" +
+						"sending us \"${e.getTag()}\" as an error. So we now have to wait a bit. Telegram\n" +
-					"asked us to wait for " + delay + " seconds.\n" +
+						"asked us to wait for ${delay} seconds.\n" +
 						"\n" +
 						"So I'm going to do just that for now. If you don't want to wait, you can quit by pressing\n" +
 						"Ctrl+C. You can restart me at any time and I will just continue to download your\n" +
 						"messages and media. But be advised that just restarting me is not going to change\n" +
-					"the fact that Telegram won't talk to me until then.")
+						"the fact that Telegram won't talk to me until then." +
-			System.out.println("")
+						"\n")
-		}
-		try {
-			TimeUnit.SECONDS.sleep(delay + 1)
-		} catch (e2: InterruptedException) {
 				}
+				hasSeenFloodWaitMessage = true
 				
+				try { TimeUnit.SECONDS.sleep(delay + 1) } catch (e: InterruptedException) { }
+			} catch (e: TimeoutException) {
+				println(
+					"\n" +
+					"Telegram took too long to respond to our request.\n" +
+					"I'm going to wait a minute and then try again." +
+					"\n")
+				try { TimeUnit.MINUTES.sleep(1) } catch (e: InterruptedException) { }
+			}
+		}
 	}
 
 	@JvmStatic
@@ -200,3 +221,5 @@ fun String.anonymize(): String {
 
 fun Any.toJson(): String = Gson().toJson(this)
 fun Any.toPrettyJson(): String = GsonBuilder().setPrettyPrinting().create().toJson(this)
+
+class MaxTriesExceededException(): RuntimeException("Max tries exceeded") {}