taskFaker.runTasks()
    assertThat(client.closed).isTrue()
    server.listener.assertExhausted() // Client should not have sent second close.
    client.listener.assertExhausted() // Server should not have sent second close.
  }

  @Test
  fun serverCloseBreaksReadMessageLoop() {
    server.webSocket!!.send("Hello!")
    server.webSocket!!.close(1000, "Bye!")
    assertThat(client.processNextFrame()).isTrue()

	if m.subroute != nil {
		msg.Flags |= FlagSubroute
	}

	// Send...
	err := m.send(msg)
	if err != nil {
		out <- Response{Err: err}
		return
	}

	// Route directly to output.
	m.respWait = out
}

// RequestStream will send a single payload request and stream back results.
// 'requests' can be nil, in which case only req is sent as input.
// It will however take less resources.

    }

    @Test
    @DisplayName("send with multiple params returns the stubbed response")
    void send_withParams_returnsResponse() throws Exception {
        // Arrange
        when(tree.send(eq(request), eq(RequestParam.NO_TIMEOUT), eq(RequestParam.NO_RETRY))).thenReturn(response);

        // Act
        CommonServerMessageBlockResponse out = tree.send(request, RequestParam.NO_TIMEOUT, RequestParam.NO_RETRY);

			// Even if there is nothing in the queue.
			if h.batchSize > 1 && time.Since(lastBatchProcess) < time.Second {
				continue
			}
		}

		// if we have reached the count send at once
		// or we have crossed last second before batch was sent, send at once
		lastBatchProcess = time.Now()

		var retries int
		retryIntvl := h.config.RetryIntvl
		if retryIntvl <= 0 {
			retryIntvl = 3 * time.Second
		}

    }

    /**
     * Get the preferred size for send operations
     *
     * @return preferred send size in bytes
     */
    public int getPreferredSendSize() {
        return preferredSendSize;
    }

    /**
     * Set the preferred size for send operations
     *
     * @param preferredSendSize preferred send size in bytes
     */

}

func (target *MQTTTarget) isActive() (bool, error) {
	if !target.client.IsConnectionOpen() {
		return false, store.ErrNotConnected
	}
	return true, nil
}

// send - sends an event to the mqtt.
func (target *MQTTTarget) send(eventData event.Event) error {
	objectName, err := url.QueryUnescape(eventData.S3.Object.Key)
	if err != nil {
		return err
	}
	key := eventData.S3.Bucket.Name + "/" + objectName

        ServerMessageBlock response = new SmbComOpenAndXResponse();

        // Execute send
        tree.send(request, response);

        // Verify session.send was called and tid was set
        verify(session).send(request, response);
        assertEquals(123, request.tid);
    }

    @Test
    void testSendWithDfs() throws SmbException {

    /**
     * @return true if the <b>debug</b> error level is enabled
     */
    boolean isDebugEnabled();

    /**
     * Send a message to the user in the <b>debug</b> error level.
     *
     * @param content
     */
    void debug(CharSequence content);

    /**
     * Send a message (and accompanying exception) to the user in the <b>debug</b> error level.<br>

        assertFalse(it.hasNext(), "Iterator should be exhausted");

        // Close explicitly to exercise doCloseInternal
        it.close();

        // Verify interactions: one create, two additional directory queries (for fetchMore), one close
        verify(tree, times(1)).send(argThat((Request<?> r) -> r instanceof Smb2CreateRequest));

        byte[] data = "Appended".getBytes();

        // When
        outputStream.write(data);

        // Then - verify write request was sent
        verify(mockTreeHandle, atLeastOnce()).send(any(Smb2WriteRequest.class), any());
    }

    @Test
    void testMultipleWrites() throws IOException, CIFSException {
        // Given