func (prod *Spooling) routeToOrigin(msg core.Message) {
	msg.Route(msg.PrevStreamID)

	if spool, exists := prod.outfile[msg.PrevStreamID]; exists {
		spool.countRead()
	}
}

func (prod *Redis) storeSortedSet(msg core.Message) {
	value, _ := prod.format.Format(msg)
	var scoreValue []byte
	if prod.fieldFromParsed {
		scoreMsg := msg
		scoreMsg.Data = value
		scoreValue, _ = prod.fieldFormat.Format(scoreMsg)
	} else {
		scoreValue, _ = prod.fieldFormat.Format(msg)
	}

	score, err := strconv.ParseFloat(string(scoreValue), 64)
	if err != nil {
		Log.Error.Print("Redis: ", err)
		return // ### return, no valid score ###
	}

	result := prod.client.ZAdd(prod.key, redis.Z{
		Score:  score,
		Member: string(value),
	})

	if result.Err() != nil {
		Log.Error.Print("Redis: ", result.Err())
		msg.Drop(prod.GetTimeout())
	}
}

func (prod *Redis) storeString(msg core.Message) {
	value, _ := prod.ProducerBase.Format(msg)

	result := prod.client.Set(prod.key, string(value))
	if result.Err() != nil {
		Log.Error.Print("Redis: ", result.Err())
		msg.Drop(prod.GetTimeout())
	}
}

func (prod *File) writeMessage(msg core.Message) {
	msg.Data, msg.StreamID = prod.ProducerBase.Format(msg)
	state, err := prod.getFileState(msg.StreamID, false)
	if err != nil {
		Log.Error.Print("File log error:", err)
		msg.Drop(time.Duration(0))
		return // ### return, dropped ###
	}

	if !state.batch.Append(msg) {
		state.writeBatch()
		state.batch.Append(msg)
	}
}

func (stream *Route) routeMessage(msg core.Message) {
	for i := 0; i < len(stream.routes); i++ {
		target := stream.routes[i]

		// Stream might require late binding
		if target.stream == nil {
			if core.StreamRegistry.WildcardProducersExist() {
				target.stream = core.StreamRegistry.GetStreamOrFallback(target.id)
			} else if target.stream = core.StreamRegistry.GetStream(target.id); target.stream == nil {
				// Remove without preserving order allows us to continue iterating
				lastIdx := len(stream.routes) - 1
				stream.routes[i] = stream.routes[lastIdx]
				stream.routes = stream.routes[:lastIdx]
				i--
				continue // ### continue, no route ###
			}
		}

		if target.id == stream.GetBoundStreamID() {
			stream.StreamBase.Route(msg, stream.GetBoundStreamID())
		} else {
			msg := msg // copy to allow streamId changes and multiple routes
			msg.StreamID = target.id
			target.stream.Enqueue(msg)
		}
	}
}

func (prod *Redis) storeHash(msg core.Message) {
	value, _ := prod.format.Format(msg)
	var field []byte
	if prod.fieldFromParsed {
		fieldMsg := msg
		fieldMsg.Data = value
		field, _ = prod.fieldFormat.Format(fieldMsg)
	} else {
		field, _ = prod.fieldFormat.Format(msg)
	}

	result := prod.client.HSet(prod.key, string(field), string(value))
	if result.Err() != nil {
		Log.Error.Print("Redis: ", result.Err())
		msg.Drop(prod.GetTimeout())
	}
}

// Format prepends the sequence number of the message (followed by ":") to the
// message.
func (format *Serialize) Format(msg core.Message) ([]byte, core.MessageStreamID) {
	msg.Data, msg.StreamID = format.base.Format(msg)
	data, err := msg.Serialize()
	if err != nil {
		Log.Error.Print("Serialize: ", err)
		return msg.Data, msg.StreamID
	}

	if !format.encode {
		return data, msg.StreamID // ### return, raw data ###
	}

	encodedSize := base64.StdEncoding.EncodedLen(len(data))
	encoded := make([]byte, encodedSize+1)
	base64.StdEncoding.Encode(encoded, data)
	encoded[encodedSize] = '\n'

	return encoded, msg.StreamID
}

func (cons *LoopBack) route(msg core.Message) {
	if streams, routeExists := cons.routes[msg.StreamID]; routeExists {
		for targetID, targetStream := range streams {
			msg.StreamID = targetID
			targetStream.Enqueue(msg)
		}
	} else {
		// Extend the cache
		targetStream := cons.addRoute(msg.StreamID, msg.StreamID)
		targetStream.Enqueue(msg)
	}
}