Method handleWriteCache

internal/nodes/api_stream.go:165–220

写入缓存

(message *pb.NodeStreamMessage)

// 写入缓存
func (this *APIStream) handleWriteCache(message *pb.NodeStreamMessage) error {
	msg := &messageconfigs.WriteCacheMessage{}
	err := json.Unmarshal(message.DataJSON, msg)
	if err != nil {
		this.replyFail(message.RequestId, "decode message data failed: "+err.Error())
		return err
	}

	storage, shouldStop, err := this.cacheStorage(message, msg.CachePolicyJSON)
	if err != nil {
		return err
	}
	if shouldStop {
		defer func() {
			storage.Stop()
		}()
	}

	expiredAt := time.Now().Unix() + msg.LifeSeconds
	writer, err := storage.OpenWriter(msg.Key, expiredAt, 200, -1, int64(len(msg.Value)), -1, false)
	if err != nil {
		this.replyFail(message.RequestId, "prepare writing failed: "+err.Error())
		return err
	}

	// 写入一个空的Header
	_, err = writer.WriteHeader([]byte(":"))
	if err != nil {
		this.replyFail(message.RequestId, "write failed: "+err.Error())
		return err
	}

	// 写入数据
	_, err = writer.Write(msg.Value)
	if err != nil {
		this.replyFail(message.RequestId, "write failed: "+err.Error())
		return err
	}

	err = writer.Close()
	if err != nil {
		this.replyFail(message.RequestId, "write failed: "+err.Error())
		return err
	}
	storage.AddToList(&caches.Item{
		Type:       writer.ItemType(),
		Key:        msg.Key,
		ExpiresAt:  expiredAt,
		HeaderSize: writer.HeaderSize(),
		BodySize:   writer.BodySize(),
	})

	this.replyOk(message.RequestId, "write ok")

	return nil
}

// 读取缓存