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

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