app/src/main/java/net/osdn/gokigen/a01d/camera/ptpip/wrapper/command/PtpIpCommandPublisher.kt

   1 package net.osdn.gokigen.a01d.camera.ptpip.wrapper.command
   2
   3 import android.util.Log
   4 import net.osdn.gokigen.a01d.camera.utils.SimpleLogDumper
   5 import java.io.BufferedReader
   6 import java.io.ByteArrayOutputStream
   7 import java.io.DataOutputStream
   8 import java.io.InputStream
   9 import java.net.InetSocketAddress
  10 import java.net.Socket
  11 import java.util.*
  12
  13 class PtpIpCommandPublisher(private val ipAddress : String, private val portNumber : Int, private val tcpNoDelay : Boolean, private val waitForever: Boolean) : IPtpIpCommandPublisher, IPtpIpCommunication
  14 {
  15     private val TAG = PtpIpCommandPublisher::class.java.simpleName
  16
  17     private val SEQUENCE_START_NUMBER = 1
  18     private val BUFFER_SIZE = 1024 * 1024 + 16 // 受信バッファは 1MB
  19
  20     private val COMMAND_SEND_RECEIVE_DURATION_MS = 5
  21     private val COMMAND_SEND_RECEIVE_DURATION_MAX = 3000
  22     private val COMMAND_POLL_QUEUE_MS = 5
  23
  24     private var isConnected = false
  25     private var isStart = false
  26     private var isHold = false
  27     private var holdId = 0
  28     private var socket : Socket? = null
  29     private var dos: DataOutputStream? = null
  30     private var bufferedReader: BufferedReader? = null
  31     private var sequenceNumber = SEQUENCE_START_NUMBER
  32     private var commandQueue: Queue<IPtpIpCommand> = ArrayDeque()
  33     private var holdCommandQueue: Queue<IPtpIpCommand> = ArrayDeque()
  34
  35     init
  36     {
  37         commandQueue.clear()
  38         holdCommandQueue.clear()
  39     }
  40
  41     override fun isConnected(): Boolean
  42     {
  43         return (isConnected)
  44     }
  45
  46     override fun connect(): Boolean
  47     {
  48         try
  49         {
  50             Log.v(TAG, " connect()")
  51             socket = Socket()
  52             socket?.tcpNoDelay = tcpNoDelay
  53             if (tcpNoDelay)
  54             {
  55                 socket?.keepAlive = false
  56                 socket?.setPerformancePreferences(0, 2, 0)
  57                 socket?.oobInline = true
  58                 socket?.reuseAddress = false
  59                 socket?.trafficClass = 0x80
  60                 //socket?.setSoLinger(true, 3000);
  61                 //socket?.setReceiveBufferSize(2097152);
  62                 //socket?.setSendBufferSize(524288);
  63             }
  64             socket?.connect(InetSocketAddress(ipAddress, portNumber), 0)
  65             isConnected = true
  66         }
  67         catch (e: Exception)
  68         {
  69             e.printStackTrace()
  70             isConnected = false
  71             socket = null
  72         }
  73         return (isConnected)
  74     }
  75
  76     override fun disconnect()
  77     {
  78         try
  79         {
  80             dos?.close()
  81             bufferedReader?.close()
  82             socket?.close()
  83             commandQueue.clear()
  84         }
  85         catch (e : Exception)
  86         {
  87             e.printStackTrace()
  88         }
  89         System.gc()
  90
  91         sequenceNumber = SEQUENCE_START_NUMBER
  92         isConnected = false
  93         isStart = false
  94         dos = null
  95         bufferedReader = null
  96         socket = null
  97     }
  98
  99     override fun start()
 100     {
 101         if (isStart)
 102         {
 103             // すでにコマンドのスレッド動作中なので抜ける
 104             return
 105         }
 106         if (socket == null)
 107         {
 108             isStart = false
 109             Log.v(TAG, " SOCKET IS NULL. (cannot start)")
 110             return
 111         }
 112
 113         isStart = true
 114         Log.v(TAG, " start()")
 115         val thread = Thread {
 116             try
 117             {
 118                 dos = DataOutputStream(socket?.getOutputStream())
 119                 while (isStart)
 120                 {
 121                     try
 122                     {
 123                         val command = commandQueue.poll()
 124                         command?.let { issueCommand(it) }
 125                         Thread.sleep(COMMAND_POLL_QUEUE_MS.toLong())
 126                     }
 127                     catch (e: Exception)
 128                     {
 129                         e.printStackTrace()
 130                     }
 131                 }
 132             }
 133             catch (e: Exception)
 134             {
 135                 Log.v(TAG, "<<<<< IP : $ipAddress port : $portNumber >>>>>")
 136                 e.printStackTrace()
 137             }
 138         }
 139         try
 140         {
 141             thread.start()
 142         }
 143         catch (e: Exception)
 144         {
 145             e.printStackTrace()
 146         }
 147     }
 148
 149     override fun stop()
 150     {
 151         isStart = false
 152         commandQueue.clear()
 153     }
 154
 155     override fun enqueueCommand(command: IPtpIpCommand): Boolean
 156     {
 157         try
 158         {
 159             if (isHold)
 160             {
 161                 return (if (holdId == command.holdId) {
 162                     if (command.isRelease)
 163                     {
 164                         // コマンドをキューに積んだ後、リリースする
 165                         val ret = commandQueue.offer(command)
 166                         isHold = false
 167
 168                         //  溜まっているキューを積みなおす
 169                         while (holdCommandQueue.size != 0)
 170                         {
 171                             val queuedCommand = holdCommandQueue.poll()
 172                             commandQueue.offer(queuedCommand)
 173                             if (queuedCommand != null && queuedCommand.isHold)
 174                             {
 175                                 // 特定シーケンスに入った場合は、そこで積みなおすのをやめる
 176                                 isHold = true
 177                                 holdId = queuedCommand.holdId
 178                                 break
 179                             }
 180                         }
 181                         return ret
 182                     }
 183                     commandQueue.offer(command)
 184                 }
 185                 else
 186                 {
 187                     // 特定シーケンスではなかったので HOLD
 188                     holdCommandQueue.offer(command)
 189                 })
 190             }
 191             if (command.isHold)
 192             {
 193                 isHold = true
 194                 holdId = command.holdId
 195             }
 196             //Log.v(TAG, "Enqueue : "  + command.getId());
 197             return (commandQueue.offer(command))
 198         }
 199         catch (e: Exception)
 200         {
 201             e.printStackTrace()
 202         }
 203         return (false)
 204     }
 205
 206     override fun flushHoldQueue(): Boolean
 207     {
 208         Log.v(TAG, "  flushHoldQueue()")
 209         holdCommandQueue.clear()
 210         System.gc()
 211         return (true)
 212     }
 213
 214     private fun issueCommand(command: IPtpIpCommand)
 215     {
 216         try
 217         {
 218             var retry_over = true
 219             while (retry_over)
 220             {
 221                 //Log.v(TAG, "issueCommand : " + command.getId());
 222                 val commandBody = command.commandBody()
 223                 if (commandBody != null)
 224                 {
 225                     // コマンドボディが入っていた場合には、コマンド送信（入っていない場合は受信待ち）
 226                     sendToCamera(command.dumpLog(), commandBody, command.useSequenceNumber(), command.embeddedSequenceNumberIndex())
 227                     val commandBody2 = command.commandBody2()
 228                     if (commandBody2 != null)
 229                     {
 230                         // コマンドボディの２つめが入っていた場合には、コマンドを連続送信する
 231                         sendToCamera(command.dumpLog(), commandBody2, command.useSequenceNumber(), command.embeddedSequenceNumberIndex2())
 232                     }
 233                     val commandBody3 = command.commandBody3()
 234                     if (commandBody3 != null)
 235                     {
 236                         // コマンドボディの３つめが入っていた場合には、コマンドを連続送信する
 237                         sendToCamera(command.dumpLog(), commandBody3, command.useSequenceNumber(), command.embeddedSequenceNumberIndex3())
 238                     }
 239                     if (command.isIncrementSeqNumber)
 240                     {
 241                         // シーケンス番号を更新する
 242                         sequenceNumber++
 243                     }
 244                 }
 245                 retry_over = receiveFromCamera(command)
 246                 if ((retry_over)&&(commandBody != null))
 247                 {
 248                     if (!command.isRetrySend)
 249                     {
 250                         while (retry_over)
 251                         {
 252                             //  コマンドを再送信しない場合はここで応答を待つ...
 253                             retry_over = receiveFromCamera(command)
 254                         }
 255                         break
 256                     }
 257                     if (!command.isIncrementSequenceNumberToRetry)
 258                     {
 259                         // 再送信...のために、シーケンス番号を戻す...
 260                         sequenceNumber--
 261                     }
 262                 }
 263             }
 264         }
 265         catch (e: Exception)
 266         {
 267             e.printStackTrace()
 268         }
 269     }
 270
 271     /**
 272      * カメラにコマンドを送信する（メイン部分）
 273      *
 274      */
 275     private fun sendToCamera(isDumpReceiveLog: Boolean, byte_array: ByteArray, useSequenceNumber: Boolean, embeddedSequenceIndex: Int)
 276     {
 277         try
 278         {
 279             if (dos == null)
 280             {
 281                 Log.v(TAG, " DataOutputStream is null.")
 282                 return
 283             }
 284
 285             // メッセージボディを加工： 最初に４バイトのレングス長をつける
 286             val sendData = ByteArray(byte_array.size + 4)
 287             sendData[0] = (byte_array.size + 4).toByte()
 288             sendData[1] = 0x00
 289             sendData[2] = 0x00
 290             sendData[3] = 0x00
 291             System.arraycopy(byte_array, 0, sendData, 4, byte_array.size)
 292             if (useSequenceNumber)
 293             {
 294                 // Sequence Number を反映させる
 295                 sendData[embeddedSequenceIndex    ] = (0x000000ff and sequenceNumber).toByte()
 296                 sendData[embeddedSequenceIndex + 1] = (0x0000ff00 and sequenceNumber ushr 8 and 0x000000ff).toByte()
 297                 sendData[embeddedSequenceIndex + 2] = (0x00ff0000 and sequenceNumber ushr 16 and 0x000000ff).toByte()
 298                 sendData[embeddedSequenceIndex + 3] = (-0x1000000 and sequenceNumber ushr 24 and 0x000000ff).toByte()
 299                 if (isDumpReceiveLog)
 300                 {
 301                     Log.v(TAG, "----- SEQ No. : $sequenceNumber -----")
 302                 }
 303             }
 304             if (isDumpReceiveLog)
 305             {
 306                 // ログに送信メッセージを出力する
 307                 SimpleLogDumper.dump_bytes("SEND[" + sendData.size + "] ", sendData)
 308             }
 309
 310             // (データを)送信
 311             dos?.write(sendData)
 312             dos?.flush()
 313         }
 314         catch (e: Exception)
 315         {
 316             e.printStackTrace()
 317         }
 318     }
 319
 320     private fun sleep(delayMs: Int)
 321     {
 322         try
 323         {
 324             Thread.sleep(delayMs.toLong())
 325         }
 326         catch (e: Exception)
 327         {
 328             e.printStackTrace()
 329         }
 330     }
 331
 332     /**
 333      * カメラからにコマンドの結果を受信する（メイン部分）
 334      *
 335      */
 336     private fun receiveFromCamera(command: IPtpIpCommand): Boolean
 337     {
 338         val callback = command.responseCallback()
 339         var delayMs = command.receiveDelayMs()
 340         if (delayMs < 0 || delayMs > COMMAND_SEND_RECEIVE_DURATION_MAX)
 341         {
 342             delayMs = COMMAND_SEND_RECEIVE_DURATION_MS
 343         }
 344
 345         return (if (callback != null && callback.isReceiveMulti)
 346         {
 347             // 受信したら逐次「受信したよ」と応答するパターン
 348             receive_multi(command, delayMs)
 349         }
 350         else
 351         {
 352             receive_single(command, delayMs)
 353         })
 354         //  受信した後、すべてをまとめて「受信したよ」と応答するパターン
 355     }
 356
 357     private fun receive_single(command: IPtpIpCommand, delayMs: Int): Boolean
 358     {
 359         val isDumpReceiveLog = command.dumpLog()
 360         val id = command.id
 361         val callback = command.responseCallback()
 362         try
 363         {
 364             val receive_message_buffer_size = BUFFER_SIZE
 365             val byte_array = ByteArray(receive_message_buffer_size)
 366             val inputStream = socket?.getInputStream()
 367             if (inputStream == null)
 368             {
 369                 Log.v(TAG, " InputStream is NULL... RECEIVE ABORTED.")
 370                 receivedAllMessage(isDumpReceiveLog, id, null, callback)
 371                 return (false)
 372             }
 373
 374             // 初回データが受信バッファにデータが溜まるまで待つ...
 375             var read_bytes = waitForReceive(inputStream, delayMs, command.maxRetryCount())
 376             if (read_bytes < 0)
 377             {
 378                 // リトライオーバー...
 379                 Log.v(TAG, " RECEIVE : RETRY OVER...")
 380                 if (!command.isRetrySend)
 381                 {
 382                     // 再送しない場合には、応答がないことを通知する
 383                     receivedAllMessage(isDumpReceiveLog, id, null, callback)
 384                 }
 385                 return (true)
 386             }
 387
 388             // 受信したデータをバッファに突っ込む
 389             val byteStream = ByteArrayOutputStream()
 390             while (read_bytes > 0)
 391             {
 392                 read_bytes = inputStream.read(byte_array, 0, receive_message_buffer_size)
 393                 if (read_bytes <= 0)
 394                 {
 395                     Log.v(TAG, " RECEIVED MESSAGE FINISHED ($read_bytes)")
 396                     break
 397                 }
 398                 byteStream.write(byte_array, 0, read_bytes)
 399                 sleep(delayMs)
 400                 read_bytes = inputStream.available()
 401             }
 402             val outputStream = cutHeader(byteStream)
 403             receivedAllMessage(isDumpReceiveLog, id, outputStream.toByteArray(), callback)
 404             System.gc()
 405         }
 406         catch (e: Throwable)
 407         {
 408             e.printStackTrace()
 409             System.gc()
 410         }
 411         return false
 412     }
 413
 414     private fun receivedAllMessage(isDumpReceiveLog: Boolean, id: Int, body: ByteArray?, callback: IPtpIpCommandCallback?)
 415     {
 416         Log.v(TAG, "receivedAllMessage() : " + (body?.size ?: 0) + " bytes.")
 417         if (isDumpReceiveLog && body != null)
 418         {
 419             // ログに受信メッセージを出力する
 420             SimpleLogDumper.dump_bytes("RECV[" + body.size + "] ", body)
 421         }
 422         callback?.receivedMessage(id, body)
 423     }
 424
 425     private fun receive_multi(command: IPtpIpCommand, delayMs: Int): Boolean
 426     {
 427         //int estimatedSize = command.estimatedReceiveDataSize();
 428         var maxRetryCount = command.maxRetryCount()
 429         val id = command.id
 430         val callback = command.responseCallback()
 431         try
 432         {
 433             Log.v(TAG, " ===== receive_multi() =====")
 434             val receive_message_buffer_size = BUFFER_SIZE
 435             val byte_array = ByteArray(receive_message_buffer_size)
 436             val inputStream = socket?.getInputStream()
 437             if (inputStream == null)
 438             {
 439                 Log.v(TAG, " InputStream is NULL... RECEIVE ABORTED.")
 440                 return (false)
 441             }
 442
 443             // 初回データが受信バッファにデータが溜まるまで待つ...
 444             var read_bytes = waitForReceive(inputStream, delayMs, command.maxRetryCount())
 445             if (read_bytes < 0)
 446             {
 447                 // リトライオーバー...
 448                 Log.v(TAG, " RECEIVE : RETRY OVER...... : " + delayMs + "ms x " + command.maxRetryCount())
 449                 if (command.isRetrySend)
 450                 {
 451                     // 要求を再送する場合、、、ダメな場合は受信待ちとする
 452                     return (true)
 453                 }
 454             }
 455             var target_length: Int
 456             var received_length: Int
 457
 458             //boolean read_retry = false;
 459             //do
 460             run {
 461
 462                 // 初回データの読み込み...
 463                 read_bytes = inputStream.read(byte_array, 0, receive_message_buffer_size)
 464                 target_length = parseDataLength(byte_array, read_bytes)
 465                 received_length = read_bytes
 466                 if (target_length <= 0)
 467                 {
 468                     // 受信サイズ異常の場合...
 469                     if (received_length > 0)
 470                     {
 471                         SimpleLogDumper.dump_bytes("WRONG DATA : ", Arrays.copyOfRange(byte_array, 0, Math.min(received_length, 64)))
 472                     }
 473                     Log.v(TAG, " WRONG LENGTH. : $target_length READ : $received_length bytes.")
 474                     callback?.receivedMessage(id, null)
 475                     return false
 476                 }
 477             } //while (read_retry);
 478
 479             Log.v(TAG, "  -=-=-=- 1st CALL : read_bytes : " + read_bytes + "(" + received_length + ") : target_length : " + target_length + "  buffer SIZE : " + byte_array.size)
 480             callback?.onReceiveProgress(received_length, target_length, Arrays.copyOfRange(byte_array, 0, received_length))
 481
 482             //do
 483             run {
 484                 sleep(delayMs)
 485                 read_bytes = inputStream.available()
 486                 if (read_bytes == 0)
 487                 {
 488                     //Log.v(TAG, " WAIT is.available() ... [" + received_length + ", " + target_length + "] retry : " + maxRetryCount);
 489                     maxRetryCount--
 490                 }
 491             } // while ((read_bytes == 0)&&(maxRetryCount > 0)&&(received_length < target_length)); // ((read_bytes == 0)&&(estimatedSize > 0)&&(received_length < estimatedSize));
 492             while (read_bytes >= 0 && received_length < target_length)
 493             {
 494                 read_bytes = inputStream.read(byte_array, 0, receive_message_buffer_size)
 495                 if (read_bytes <= 0)
 496                 {
 497                     Log.v(TAG, "  RECEIVED MESSAGE FINISHED ($read_bytes)")
 498                     break
 499                 }
 500                 received_length = received_length + read_bytes
 501
 502                 //  一時的な処理
 503                 callback?.onReceiveProgress(received_length, target_length, Arrays.copyOfRange(byte_array, 0, read_bytes))
 504                 //byteStream.write(byte_array, 0, read_bytes);
 505                 maxRetryCount = command.maxRetryCount()
 506                 //do
 507                 run {
 508                     sleep(delayMs)
 509                     read_bytes = inputStream.available()
 510                     //Log.v(TAG, "  is.available() read_bytes : " + read_bytes + " " + received_length + " < " + estimatedSize);
 511                     if (read_bytes == 0) {
 512                         Log.v(TAG, " WAIT is.available() ... [$received_length, $target_length] $read_bytes retry : $maxRetryCount")
 513                         maxRetryCount--
 514                     }
 515                 } // while ((read_bytes == 0)&&(maxRetryCount > 0)&&(received_length < target_length)); // while ((read_bytes == 0)&&(estimatedSize > 0)&&(received_length < estimatedSize));
 516             }
 517
 518             //  終了報告...一時的？
 519             Log.v(TAG, "  --- receive_multi : $id  ($read_bytes) [$maxRetryCount] $receive_message_buffer_size ($received_length) ")
 520             callback?.receivedMessage(id, Arrays.copyOfRange(byte_array, 0, received_length))
 521         }
 522         catch (e: Throwable)
 523         {
 524             e.printStackTrace()
 525         }
 526         return false
 527     }
 528
 529     private fun parseDataLength(byte_array: ByteArray, read_bytes: Int): Int
 530     {
 531         var offset = 0
 532         var lenlen = 0
 533         //int packetType = 0;
 534         try
 535         {
 536             if (read_bytes > 20)
 537             {
 538                 if (byte_array[offset + 4].toUByte().toInt() == 0x07)
 539                 {
 540                     // 前の応答が入っていると考える...
 541                     offset = 14
 542                 }
 543                 if (byte_array[offset + 4].toUByte().toInt() == 0x09)
 544                 {
 545                     lenlen = (byte_array[offset + 15].toUByte().toInt() and 0xff shl 24) + (byte_array[offset + 14].toUByte().toInt() and 0xff shl 16) + (byte_array[offset + 13].toUByte().toInt() and 0xff shl 8) + (byte_array[offset + 12].toUByte().toInt() and 0xff)
 546                     //packetType = (((int)byte_array[offset + 16]) & 0xff);
 547                 }
 548             }
 549             //Log.v(TAG, " --- parseDataLength() length: " + lenlen + " TYPE: " + packetType + " read_bytes: " + read_bytes + "  offset : " + offset);
 550         }
 551         catch (e: Exception)
 552         {
 553             e.printStackTrace()
 554         }
 555         return (lenlen)
 556     }
 557
 558     private fun cutHeader(receivedBuffer: ByteArrayOutputStream): ByteArrayOutputStream
 559     {
 560         try
 561         {
 562             val byte_array = receivedBuffer.toByteArray()
 563             val limit = byte_array.size
 564             var lenlen = 0
 565             val len = (byte_array[3].toUByte().toInt() and 0xff shl 24) + (byte_array[2].toUByte().toInt() and 0xff shl 16) + (byte_array[1].toUByte().toInt() and 0xff shl 8) + (byte_array[0].toUByte().toInt() and 0xff)
 566             val packetType = byte_array[4].toUByte().toInt() and 0xff
 567             if (limit == len || limit < 16384)
 568             {
 569                 // 応答は１つしか入っていない。もしくは受信データサイズが16kBの場合は、そのまま返す。
 570                 return (receivedBuffer)
 571             }
 572             if (packetType == 0x09)
 573             {
 574                 lenlen = (byte_array[15].toUByte().toInt() and 0xff shl 24) + (byte_array[14].toUByte().toInt() and 0xff shl 16) + (byte_array[13].toUByte().toInt() and 0xff shl 8) + (byte_array[12].toUByte().toInt() and 0xff)
 575                 //packetType = (((int) byte_array[16]) & 0xff);
 576             }
 577             // Log.v(TAG, " ---  RECEIVED MESSAGE : " + len + " bytes (BUFFER: " + byte_array.length + " bytes)" + " length : " + lenlen + " TYPE : " + packetType + " --- ");
 578             if (lenlen == 0)
 579             {
 580                 // データとしては変なので、なにもしない
 581                 return receivedBuffer
 582             }
 583             val outputStream = ByteArrayOutputStream()
 584             //outputStream.write(byte_array, 0, 20);  //
 585             var position = 20 // ヘッダ込の先頭
 586             while (position < limit)
 587             {
 588                 lenlen = (byte_array[position + 3].toUByte().toInt() and 0xff shl 24) + (byte_array[position + 2].toUByte().toInt() and 0xff shl 16) + (byte_array[position + 1].toUByte().toInt() and 0xff shl 8) + (byte_array[position].toUByte().toInt() and 0xff)
 589
 590                 val copyByte = Math.min(limit - (position + 12), lenlen - 12)
 591                 outputStream.write(byte_array, position + 12, copyByte)
 592                 position = position + lenlen
 593             }
 594             return (outputStream)
 595         }
 596         catch (e: Throwable)
 597         {
 598             e.printStackTrace()
 599             System.gc()
 600         }
 601         return (receivedBuffer)
 602     }
 603
 604     private fun waitForReceive(inputStream : InputStream, delayMs: Int, retryCount : Int): Int
 605     {
 606         var retry_count = retryCount
 607         var isLogOutput = true
 608         var read_bytes = 0
 609         try
 610         {
 611             while (read_bytes <= 0)
 612             {
 613                 sleep(delayMs)
 614                 read_bytes = inputStream.available()
 615                 if (read_bytes <= 0)
 616                 {
 617                     if (isLogOutput)
 618                     {
 619                         Log.v(TAG, "waitForReceive:: is.available() WAIT... : " + delayMs + "ms")
 620                         isLogOutput = false
 621                     }
 622                     retry_count--
 623                     if (!waitForever && retry_count < 0)
 624                     {
 625                         return -1
 626                     }
 627                 }
 628             }
 629         }
 630         catch (e: Exception)
 631         {
 632             e.printStackTrace()
 633         }
 634         return read_bytes
 635     }
 636 }