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