}
			}
		case *googleapi.Error:
			apiErr = APIError{
				Code:           "XGCSInternalError",
				Description:    e.Message,
				HTTPStatusCode: e.Code,
			}
			// GCS may send multiple errors, just pick the first one
			// since S3 only sends one Error XML response.
			if len(e.Errors) >= 1 {
				apiErr.Code = e.Errors[0].Reason
			}
		case azblob.StorageError:
			apiErr = APIError{

	c.Assert(response.Header.Get("Content-Type"), "application/json")
}

// TestPartialContent - Validating for GetObject with partial content request.
// By setting the Range header, A request to send specific bytes range of data from an
// already uploaded object can be done.
func (s *TestSuiteCommon) TestPartialContent(c *check) {
	bucketName := getRandomBucketName()

		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, defaultSingleRequestTimeout)
		defer cancel()
	}
	// Send request
	if err := m.send(msg); err != nil {
		return nil, err
	}
	if debugReqs {
		fmt.Println(m.MuxID, m.parent.String(), "SEND")
	}
	// Wait for response or context.
	select {
	case v, ok := <-ch:
		if !ok {
			return nil, ErrDisconnected
		}

	CurrentSendCalls int64 // CurrentSendCalls is the number of concurrent async Send calls to all targets
	CurrentQueue     int   // Populated if target has a store.
	TotalEvents      int64
	FailedEvents     int64 // Number of failed events per target
}

// TargetList - holds list of targets indexed by target ID.
type TargetList struct {
	// The number of concurrent async Send calls to all targets
	currentSendCalls  atomic.Int64

}

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

		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)

							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())

func (c *Target) Endpoint() string {
	return ""
}

func (c *Target) String() string {
	return "console"
}

// Send log message 'e' to console
func (c *Target) Send(e interface{}) error {
	entry, ok := e.(log.Entry)
	if !ok {
		return fmt.Errorf("Uexpected log entry structure %#v", e)
	}
	if logger.IsJSON() {
		logJSON, err := json.Marshal(&entry)
		if err != nil {
			return err
		}

		if err := ch.Confirm(false); err != nil {
			ch.Close()
			return nil, nil, err
		}
		return ch, confirms, nil
	}

	return ch, nil, nil
}

// send - sends an event to the AMQP091.
func (target *AMQPTarget) send(eventData event.Event, ch *amqp091.Channel, confirms chan amqp091.Confirmation) error {
	objectName, err := url.QueryUnescape(eventData.S3.Object.Key)
	if err != nil {
		return err
	}

	// Requests sent cannot be used any further by the called.
	Requests chan<- []byte

	muxID uint64
	ctx   context.Context
}

// Send a payload to the remote server.
func (s *Stream) Send(b []byte) error {
	if s.Requests == nil {
		return errors.New("stream does not accept requests")
	}
	select {
	case s.Requests <- b:
		return nil
	case <-s.ctx.Done():