View Javadoc
1   /*
2    * ====================================================================
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *   http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing,
14   * software distributed under the License is distributed on an
15   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16   * KIND, either express or implied.  See the License for the
17   * specific language governing permissions and limitations
18   * under the License.
19   * ====================================================================
20   *
21   * This software consists of voluntary contributions made by many
22   * individuals on behalf of the Apache Software Foundation.  For more
23   * information on the Apache Software Foundation, please see
24   * <http://www.apache.org/>.
25   *
26   */
27  package org.apache.hc.core5.http.nio.entity;
28  
29  import java.io.IOException;
30  import java.nio.ByteBuffer;
31  import java.util.Set;
32  
33  import org.apache.hc.core5.http.ContentType;
34  import org.apache.hc.core5.http.nio.AsyncEntityProducer;
35  import org.apache.hc.core5.http.nio.DataStreamChannel;
36  import org.apache.hc.core5.http.nio.StreamChannel;
37  import org.apache.hc.core5.util.Args;
38  
39  /**
40   * @since 5.0
41   */
42  public abstract class AbstractBinAsyncEntityProducer implements AsyncEntityProducer {
43  
44      private final ByteBuffer bytebuf;
45      private final int fragmentSizeHint;
46      private final ContentType contentType;
47  
48      private volatile boolean endStream;
49  
50      public AbstractBinAsyncEntityProducer(
51              final int bufferSize,
52              final int fragmentSizeHint,
53              final ContentType contentType) {
54          Args.positive(bufferSize, "Buffer size");
55          this.bytebuf = ByteBuffer.allocate(bufferSize);
56          this.fragmentSizeHint = fragmentSizeHint >= 0 ? fragmentSizeHint : bufferSize / 2;
57          this.contentType = contentType;
58      }
59  
60      protected abstract void produceData(StreamChannel<ByteBuffer> channel) throws IOException;
61  
62      @Override
63      public final String getContentType() {
64          return contentType != null ? contentType.toString() : null;
65      }
66  
67      @Override
68      public String getContentEncoding() {
69          return null;
70      }
71  
72      @Override
73      public boolean isChunked() {
74          return false;
75      }
76  
77      @Override
78      public Set<String> getTrailerNames() {
79          return null;
80      }
81  
82      @Override
83      public final void produce(final DataStreamChannel channel) throws IOException {
84          produceData(new StreamChannel<ByteBuffer>() {
85  
86              @Override
87              public int write(final ByteBuffer src) throws IOException {
88                  Args.notNull(src, "Buffer");
89                  final int chunk = src.remaining();
90                  if (chunk == 0) {
91                      return 0;
92                  }
93                  if (bytebuf.remaining() >= chunk) {
94                      bytebuf.put(src);
95                      return chunk;
96                  }
97                  int totalBytesWritten = 0;
98                  if (!bytebuf.hasRemaining() || bytebuf.position() >= fragmentSizeHint) {
99                      bytebuf.flip();
100                     final int bytesWritten = channel.write(bytebuf);
101                     bytebuf.compact();
102                     totalBytesWritten += bytesWritten;
103                 }
104                 if (bytebuf.position() == 0) {
105                     final int bytesWritten = channel.write(src);
106                     totalBytesWritten += bytesWritten;
107                 }
108                 return totalBytesWritten;
109             }
110 
111             @Override
112             public void endStream() throws IOException {
113                 endStream = true;
114             }
115 
116         });
117 
118         if (endStream || !bytebuf.hasRemaining() || bytebuf.position() >= fragmentSizeHint) {
119             bytebuf.flip();
120             channel.write(bytebuf);
121             bytebuf.compact();
122         }
123         if (bytebuf.position() == 0 && endStream) {
124             channel.endStream();
125         }
126     }
127 
128 }