h.logEntry(entry)
	}
}

func (h *Target) logEntry(entry interface{}) {
	atomic.AddInt64(&h.totalMessages, 1)
	if err := h.send(entry); err != nil {
		atomic.AddInt64(&h.failedMessages, 1)
		h.kconfig.LogOnce(context.Background(), err, h.kconfig.Topic)
	}
}

func (h *Target) send(entry interface{}) error {
	if err := h.initKafkaOnce.Do(h.init); err != nil {
		return err
	}
	logJSON, err := json.Marshal(&entry)

    taskFaker.runTasks()
    client.listener.assertFailure(IOException::class.java, "source is closed")
    assertThat(client.webSocket!!.send("Hello!")).isFalse()
  }

  @Test
  fun socketClosedDuringMessageKillsWebSocket() {
    client2Server.source.close()
    assertThat(client.webSocket!!.send("Hello!")).isTrue()
    taskFaker.runTasks()
    client.listener.assertFailure(IOException::class.java, "source is closed")

			t.Errorf("failed to read from client. %v", terr)
			return
		}
		received = string(b)
		if received != "message two\n" {
			t.Errorf(`server: expected: "message two\n", got: %v`, received)
			return
		}

		// Send a response.
		_, terr = io.WriteString(deadlineconn, "messages received\n")
		if terr != nil {
			t.Errorf("failed to write to client. %v", terr)
			return
		}
	}()

	c, err := net.Dial("tcp", serverAddr)

### Conditional Cache Hit
 
When cache flags require checking the cache results are still valid an early cacheConditionalHit event is
received followed by a cache hit or miss.  Critically in the cache hit scenario the server won’t send the response body.

The response will have non-null `cacheResponse` and `networkResponse`. The cacheResponse will be used as the top level
response only if the response code is HTTP/1.1 304 Not Modified.
 
 - CallStart

JwIDAQABMA0GCSqGSIb3DQEBBQUAA4GBALqjOA6bD8BEl7hkQ8XwX/owSAL0URDe
nUfCOsXgIIAqgw4uTCLOfCJVZNKmRT+KguvPAQ6Z80vau2UxPX5Q2Q+OHXDRrEnK
FjqSBgLP06Qw7a++bshlWGTt5bHWOneW3EQikedckVuIKPkOCib9yGi4VmBBjdFE
M9ofSEt/bdRD
-----END CERTIFICATE-----`)
	if err != nil {
		t.Fatalf("Unable to create temporary file. %v", err)
	}
	defer os.Remove(tempFile2)

	tempFile3, err := createTempFile("public-cert-file",
		`-----BEGIN CERTIFICATE-----

    Project: $project.name
    #if ( $project.url )Project URL: ${project.url}#end

    License: $license.name#if ( $spdx ) ($spdx)#end

    License URL: $license.url ($licFile)

#*  *##end
#**##end

							dst = (dst + 1) % len(managers)
						}
						local := managers[src]
						conn := local.Connection(hosts[dst])
						if conn == nil {
							b.Fatal("No connection")
						}
						// Send the payload.
						t := time.Now()
						resp, err := conn.Request(ctx, handlerTest, payload)
						latency += time.Since(t).Nanoseconds()
						if err != nil {
							if debugReqs {
								fmt.Println(err.Error())

	if err != nil {
		return err
	}
	if msg.Flags&FlagCRCxxh3 != 0 {
		h := xxh3.Hash(dst)
		dst = binary.LittleEndian.AppendUint32(dst, uint32(h))
	}
	return c.send(c.ctx, dst)
}

// sendMsg will send
func (c *Connection) sendMsg(conn net.Conn, msg message, payload msgp.MarshalSizer) error {
	if payload != nil {
		if sz := payload.Msgsize(); cap(msg.Payload) < sz {
			PutByteBuffer(msg.Payload)

 * multiplexing requests. It encapsulates a stream and some protocol
 * knowledge (provided by a concrete subclass) so that connecting,
 * disconnecting, sending, and receiving can be syncronized
 * properly. Apparatus is provided to send and receive requests
 * concurrently.
 */

public abstract class Transport implements Runnable, AutoCloseable {

    private static int id = 0;

  private final int end; // exclusive

  private ImmutableLongArray(long[] array) {
    this(array, 0, array.length);
  }

  private ImmutableLongArray(long[] array, int start, int end) {
    this.array = array;
    this.start = start;
    this.end = end;
  }

  /** Returns the number of values in this array. */
  public int length() {