End HTTP/2 stream with empty DATA rather than empty HEADERS

When ending the stream without any payload, use a DATA frame rather than
a HEADERS frame to work around golang/go#47851.

Tested with https://github.com/raboof/http2-from-go-to-akka-http

Author: raboof

akka-http-core/src/main/scala/akka/http/impl/engine/http2/hpack/HeaderCompression.scala

@@ -32,38 +32,44 @@ private[http2] object HeaderCompression extends GraphStage[FlowShape[FrameEvent,
     val encoder = new akka.http.shaded.com.twitter.hpack.Encoder(Http2Protocol.InitialMaxHeaderTableSize)
     val os = new ByteArrayOutputStream(128)
 
+<<<<<<< HEAD
     def onPull(): Unit = pull(eventsIn)
     def onPush(): Unit = grab(eventsIn) match {
       case ack @ SettingsAckFrame(s) =>
         applySettings(s)
         push(eventsOut, ack)
       case ParsedHeadersFrame(streamId, endStream, kvs, prioInfo) =>
-        kvs.foreach {
-          case (key, value: String) =>
-            encoder.encodeHeader(os, key, value, false)
-          case (key, value) =>
-            throw new IllegalStateException(s"Didn't expect key-value-pair [$key] -> [$value](${value.getClass}) here.")
-        }
-        val result = ByteString.fromArrayUnsafe(os.toByteArray) // BAOS.toByteArray always creates a copy
-        os.reset()
-        if (result.size <= currentMaxFrameSize) push(eventsOut, HeadersFrame(streamId, endStream, endHeaders = true, result, prioInfo))
+        // When ending the stream without any payload, use a DATA frame rather than
+        // a HEADERS frame to work around https://github.com/golang/go/issues/47851.
+        if (endStream && kvs.isEmpty) push(eventsOut, DataFrame(streamId, endStream, ByteString.empty))
         else {
-          val first = HeadersFrame(streamId, endStream, endHeaders = false, result.take(currentMaxFrameSize), prioInfo)
+          kvs.foreach {
+            case (key, value: String) =>
+              encoder.encodeHeader(os, key, value, false)
+            case (key, value) =>
+              throw new IllegalStateException(s"Didn't expect key-value-pair [$key] -> [$value](${value.getClass}) here.")
+          }
+          val result = ByteString.fromArrayUnsafe(os.toByteArray) // BAOS.toByteArray always creates a copy
+          os.reset()
+          if (result.size <= currentMaxFrameSize) push(eventsOut, HeadersFrame(streamId, endStream, endHeaders = true, result, prioInfo))
+          else {
+            val first = HeadersFrame(streamId, endStream, endHeaders = false, result.take(currentMaxFrameSize), prioInfo)
 
-          push(eventsOut, first)
-          setHandler(eventsOut, new OutHandler {
-            private var remainingData = result.drop(currentMaxFrameSize)
+            push(eventsOut, first)
+            setHandler(eventsOut, new OutHandler {
+              private var remainingData = result.drop(currentMaxFrameSize)
 
-            def onPull(): Unit = {
-              val thisFragment = remainingData.take(currentMaxFrameSize)
-              val rest = remainingData.drop(currentMaxFrameSize)
-              val last = rest.isEmpty
+              def onPull(): Unit = {
+                val thisFragment = remainingData.take(currentMaxFrameSize)
+                val rest = remainingData.drop(currentMaxFrameSize)
+                val last = rest.isEmpty
 
-              push(eventsOut, ContinuationFrame(streamId, endHeaders = last, thisFragment))
-              if (last) setHandler(eventsOut, logic)
-              else remainingData = rest
-            }
-          })
+                push(eventsOut, ContinuationFrame(streamId, endHeaders = last, thisFragment))
+                if (last) setHandler(eventsOut, logic)
+                else remainingData = rest
+              }
+            })
+          }
         }
       case x => push(eventsOut, x)
     }