diff --git a/api/config/config.go b/api/config/config.go index fef1580..0f8304b 100644 --- a/api/config/config.go +++ b/api/config/config.go @@ -23,6 +23,7 @@ const ( func ParseFlags() ServerConfig { host := flag.String("Host", "localhost", "Hostname") port := flag.Int("Port", 8081, "Listen port") + rntbdPort := flag.Int("RntbdPort", 10000, "RNTBD listen port") explorerPath := flag.String("ExplorerDir", "", "Path to cosmos-explorer files") tlsCertificatePath := flag.String("Cert", "", "Hostname") tlsCertificateKey := flag.String("CertKey", "", "Hostname") @@ -35,6 +36,7 @@ func ParseFlags() ServerConfig { flag.Var(logLevel, "LogLevel", fmt.Sprintf("Sets the logging level %s", logLevel.AllowedValuesList())) dataStore := NewEnumValue("json", []string{DataStoreJson, DataStoreBadger}) flag.Var(dataStore, "DataStore", fmt.Sprintf("Sets the data store %s", dataStore.AllowedValuesList())) + enableRntbd := flag.Bool("ExperimentalEnableRntbd", false, "EXPERIMENTAL: Enable RNTBD (CosmosDB Direct Connection Mode)") flag.Parse() setFlagsFromEnvironment() @@ -42,6 +44,7 @@ func ParseFlags() ServerConfig { config := ServerConfig{} config.Host = *host config.Port = *port + config.RntbdPort = *rntbdPort config.ExplorerPath = *explorerPath config.TLS_CertificatePath = *tlsCertificatePath config.TLS_CertificateKey = *tlsCertificateKey @@ -52,6 +55,7 @@ func ParseFlags() ServerConfig { config.AccountKey = *accountKey config.LogLevel = logLevel.value config.DataStore = dataStore.value + config.EnableRntbd = *enableRntbd config.PopulateCalculatedFields() @@ -62,6 +66,7 @@ func (c *ServerConfig) PopulateCalculatedFields() { c.DatabaseAccount = c.Host c.DatabaseDomain = c.Host c.DatabaseEndpoint = fmt.Sprintf("https://%s:%d/", c.Host, c.Port) + c.RntbdEndpoint = fmt.Sprintf("rntbd://%s:%d/", c.Host, c.RntbdPort) c.ExplorerBaseUrlLocation = ExplorerBaseUrlLocation switch c.LogLevel { diff --git a/api/config/models.go b/api/config/models.go index 4613160..83b905d 100644 --- a/api/config/models.go +++ b/api/config/models.go @@ -4,10 +4,12 @@ type ServerConfig struct { DatabaseAccount string `json:"databaseAccount"` DatabaseDomain string `json:"databaseDomain"` DatabaseEndpoint string `json:"databaseEndpoint"` + RntbdEndpoint string `json:"rntbdEndpoint"` AccountKey string `json:"accountKey"` ExplorerPath string `json:"explorerPath"` Port int `json:"port"` + RntbdPort int `json:"rntbdPort"` Host string `json:"host"` TLS_CertificatePath string `json:"tlsCertificatePath"` TLS_CertificateKey string `json:"tlsCertificateKey"` @@ -17,6 +19,7 @@ type ServerConfig struct { DisableTls bool `json:"disableTls"` LogLevel string `json:"logLevel"` ExplorerBaseUrlLocation string `json:"explorerBaseUrlLocation"` + EnableRntbd bool `json:"enableRntbd"` DataStore string `json:"dataStore"` } diff --git a/api/handlers/server_info.go b/api/handlers/server_info.go index 3fb0487..ac09a95 100644 --- a/api/handlers/server_info.go +++ b/api/handlers/server_info.go @@ -3,6 +3,7 @@ package handlers import ( "fmt" "net/http" + "strings" "github.com/gin-gonic/gin" ) @@ -41,3 +42,46 @@ func (h *Handlers) GetServerInfo(c *gin.Context) { "queryEngineConfiguration": "{\"allowNewKeywords\":true,\"maxJoinsPerSqlQuery\":10,\"maxQueryRequestTimeoutFraction\":0.9,\"maxSqlQueryInputLength\":524288,\"maxUdfRefPerSqlQuery\":10,\"queryMaxInMemorySortDocumentCount\":-1000,\"spatialMaxGeometryPointCount\":256,\"sqlAllowNonFiniteNumbers\":false,\"sqlDisableOptimizationFlags\":0,\"enableSpatialIndexing\":true,\"maxInExpressionItemsCount\":2147483647,\"maxLogicalAndPerSqlQuery\":2147483647,\"maxLogicalOrPerSqlQuery\":2147483647,\"maxSpatialQueryCells\":2147483647,\"sqlAllowAggregateFunctions\":true,\"sqlAllowGroupByClause\":true,\"sqlAllowLike\":true,\"sqlAllowSubQuery\":true,\"sqlAllowScalarSubQuery\":true,\"sqlAllowTop\":true}", }) } + +type Address struct { + IsPrimary bool `json:"isPrimary"` + PhyscialUri string `json:"physcialUri"` + IsAuxiliary bool `json:"isAuxiliary"` + PartitionTargetReplicaSetSize int `json:"partitionTargetReplicaSetSize"` + Protocol string `json:"protocol"` + PartitionKeyRangeId string `json:"partitionKeyRangeId"` + PartitionIndex string `json:"partitionIndex"` +} + +func (h *Handlers) GetAddresses(c *gin.Context) { + addresses := []Address{} + + if h.config.EnableRntbd { + addresses = append(addresses, Address{ + IsPrimary: true, + PhyscialUri: h.config.RntbdEndpoint, + IsAuxiliary: false, + PartitionTargetReplicaSetSize: 1, + Protocol: "rntbd", + PartitionKeyRangeId: "0", + PartitionIndex: "0@0", + }) + } + + if !strings.Contains(c.Request.RequestURI, "protocol%20eq%20rntbd") { + addresses = append(addresses, Address{ + IsPrimary: true, + PhyscialUri: h.config.DatabaseEndpoint, + IsAuxiliary: false, + PartitionTargetReplicaSetSize: 1, + Protocol: "https", + PartitionKeyRangeId: "0", + PartitionIndex: "0@0", + }) + } + + c.IndentedJSON(http.StatusOK, gin.H{ + "Addresss": addresses, + "_count": len(addresses), + }) +} diff --git a/api/headers/headers.go b/api/headers/headers.go index 92c9e26..4f73672 100644 --- a/api/headers/headers.go +++ b/api/headers/headers.go @@ -21,4 +21,10 @@ const ( // Kinda retarded, but what can I do ¯\_(ツ)_/¯ IsQuery = "x-ms-documentdb-isquery" // Sent from python sdk and web explorer Query = "x-ms-documentdb-query" // Sent from Go sdk + + // I kinda don't use these, but I've seen them in the wild xd + SupportedCapabilities = "x-ms-cosmos-sdk-supportedcapabilities" + ClientRetryAttemptCount = "x-ms-client-retry-attempt-count" + RemainingTimeInMsOnClient = "x-ms-remaining-time-in-ms-on-client" + ConsistencyLevel = "x-ms-consistency-level" ) diff --git a/api/router.go b/api/router.go index 4691a0f..da19370 100644 --- a/api/router.go +++ b/api/router.go @@ -31,6 +31,7 @@ func (s *ApiServer) CreateRouter(dataStore datastore.DataStore) { router := gin.Default(func(e *gin.Engine) { e.RedirectTrailingSlash = false + e.RemoveExtraSlash = true }) if s.config.LogLevel == "debug" { @@ -79,6 +80,7 @@ func (s *ApiServer) CreateRouter(dataStore datastore.DataStore) { router.GET("/offers", handlers.GetOffers) router.GET("/", routeHandlers.GetServerInfo) + router.GET("//addresses", routeHandlers.GetAddresses) router.GET("/cosmium/export", routeHandlers.CosmiumExport) diff --git a/cmd/rntbd_parser/rntbd_parser.go b/cmd/rntbd_parser/rntbd_parser.go new file mode 100644 index 0000000..85f9db2 --- /dev/null +++ b/cmd/rntbd_parser/rntbd_parser.go @@ -0,0 +1,63 @@ +package main + +import ( + "encoding/hex" + "encoding/json" + "flag" + "fmt" + + "github.com/pikami/cosmium/internal/rntbd" +) + +func main() { + input := flag.String("input", "", "Input hex string") + isResponse := flag.Bool("response", false, "Is response") + flag.Parse() + + data, err := hex.DecodeString(*input) + if err != nil { + fmt.Printf("Error decoding hex string: %v\n", err) + return + } + + frame, err := rntbd.ParseFrame(data, *isResponse) + if err != nil { + fmt.Printf("Error parsing frame: %v\n", err) + return + } + + fmt.Printf("Activity ID: %s\n", hex.EncodeToString(frame.ActivityId)) + fmt.Printf("Resource Type: %s\n", frame.ResourceType.String()) + fmt.Printf("Operation Type: %s\n", frame.OperationType.String()) + + if len(frame.RequestHeaders) > 0 { + fmt.Printf("=== Request Headers ===\n") + for header, value := range frame.RequestHeaders { + fmt.Printf("%s: %v\n", header.String(), value) + } + } + + if len(frame.ResponseHeaders) > 0 { + fmt.Printf("=== Response Headers ===\n") + for header, value := range frame.ResponseHeaders { + fmt.Printf("%s: %v\n", header.String(), value) + } + } + + if len(frame.ContextHeaders) > 0 { + fmt.Printf("=== Context Headers ===\n") + for header, value := range frame.ContextHeaders { + fmt.Printf("%s: %v\n", header.String(), value) + } + } + + if len(frame.Payload) > 0 { + var jsonObj any + err := json.Unmarshal(frame.Payload, &jsonObj) + if err != nil { + fmt.Printf("Payload: %s\n", hex.EncodeToString(frame.Payload)) + } else { + fmt.Printf("Payload: %+v\n", jsonObj) + } + } +} diff --git a/cmd/server/server.go b/cmd/server/server.go index 207151d..d66222c 100644 --- a/cmd/server/server.go +++ b/cmd/server/server.go @@ -11,6 +11,7 @@ import ( badgerdatastore "github.com/pikami/cosmium/internal/datastore/badger_datastore" jsondatastore "github.com/pikami/cosmium/internal/datastore/json_datastore" "github.com/pikami/cosmium/internal/logger" + "github.com/pikami/cosmium/internal/rntbd" ) func main() { @@ -38,6 +39,15 @@ func main() { panic(err) } + if configuration.EnableRntbd { + rntbdServer := rntbd.NewRntbdServer(configuration.RntbdPort, server) + err = rntbdServer.Start() + if err != nil { + panic(err) + } + defer rntbdServer.Stop() + } + waitForExit(server, dataStore) } @@ -50,6 +60,5 @@ func waitForExit(server *api.ApiServer, dataStore datastore.DataStore) { // Stop the server server.Stop() - dataStore.Close() } diff --git a/internal/rntbd/builder.go b/internal/rntbd/builder.go new file mode 100644 index 0000000..4f0149d --- /dev/null +++ b/internal/rntbd/builder.go @@ -0,0 +1,138 @@ +package rntbd + +import ( + "bytes" + "encoding/binary" +) + +type RntbdResponseFrame struct { + StatusCode uint16 + ResourceType RntbdResourceType + ActivityId []byte + ResponseHeaders []RntbdResponseHeader + Payload []byte +} + +type RntbdResponseHeader struct { + HeaderId uint16 + TokenType RntbdTokenType + TokenValue any +} + +type RntbdResponseFrameBuilder struct { + frame RntbdResponseFrame +} + +func (b *RntbdResponseFrameBuilder) AddHeader(headerId uint16, tokenType RntbdTokenType, tokenValue any) { + b.frame.ResponseHeaders = append(b.frame.ResponseHeaders, RntbdResponseHeader{ + HeaderId: headerId, + TokenType: tokenType, + TokenValue: tokenValue, + }) +} + +func (b *RntbdResponseFrameBuilder) AddPayload(payload []byte) { + b.frame.Payload = payload +} + +func (b *RntbdResponseFrameBuilder) SetStatusCode(statusCode uint16) { + b.frame.StatusCode = statusCode +} + +func (b *RntbdResponseFrameBuilder) SetResourceType(resourceType RntbdResourceType) { + b.frame.ResourceType = resourceType +} + +func (b *RntbdResponseFrameBuilder) SetActivityId(activityId []byte) { + b.frame.ActivityId = activityId +} + +func (b *RntbdResponseFrameBuilder) Build() *RntbdResponseFrame { + return &b.frame +} + +func (f *RntbdResponseFrame) ToBytes() []byte { + var buffer bytes.Buffer + + binary.Write(&buffer, binary.LittleEndian, f.StatusCode) + binary.Write(&buffer, binary.LittleEndian, uint16(f.ResourceType)) + binary.Write(&buffer, binary.LittleEndian, f.ActivityId) + + for _, header := range f.ResponseHeaders { + binary.Write(&buffer, binary.LittleEndian, header.HeaderId) + binary.Write(&buffer, binary.LittleEndian, uint8(header.TokenType)) + + switch header.TokenType { + case RntbdTokenTypeByte: + buffer.Write(header.TokenValue.([]byte)) + case RntbdTokenTypeUShort: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(uint16)) + case RntbdTokenTypeULong: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(uint32)) + case RntbdTokenTypeLong: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(int32)) + case RntbdTokenTypeULongLong: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(uint64)) + case RntbdTokenTypeLongLong: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(int64)) + case RntbdTokenTypeGuid: + buffer.Write(header.TokenValue.([]byte)) + + case RntbdTokenTypeSmallString: + binary.Write(&buffer, binary.LittleEndian, uint8(len(header.TokenValue.(string)))) + buffer.WriteString(header.TokenValue.(string)) + case RntbdTokenTypeString: + binary.Write(&buffer, binary.LittleEndian, uint16(len(header.TokenValue.(string)))) + buffer.WriteString(header.TokenValue.(string)) + case RntbdTokenTypeULongString: + binary.Write(&buffer, binary.LittleEndian, uint32(len(header.TokenValue.(string)))) + buffer.WriteString(header.TokenValue.(string)) + case RntbdTokenTypeSmallBytes: + binary.Write(&buffer, binary.LittleEndian, uint8(len(header.TokenValue.([]byte)))) + buffer.Write(header.TokenValue.([]byte)) + case RntbdTokenTypeBytes: + binary.Write(&buffer, binary.LittleEndian, uint16(len(header.TokenValue.([]byte)))) + buffer.Write(header.TokenValue.([]byte)) + case RntbdTokenTypeULongBytes: + binary.Write(&buffer, binary.LittleEndian, uint32(len(header.TokenValue.([]byte)))) + buffer.Write(header.TokenValue.([]byte)) + case RntbdTokenTypeFloat: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(float32)) + case RntbdTokenTypeDouble: + binary.Write(&buffer, binary.LittleEndian, header.TokenValue.(float64)) + case RntbdTokenTypeInvalid: + panic("invalid token type") + default: + panic("invalid token type") + } + } + + payloadSize := uint32(0) + if len(f.Payload) > 0 { + payloadSize = uint32(len(f.Payload)) + 4 + } + + frameSize := uint32(buffer.Len()) + 4 + result := make([]byte, frameSize+payloadSize) + binary.LittleEndian.PutUint32(result, frameSize) + copy(result[4:], buffer.Bytes()) + + if len(f.Payload) > 0 { + binary.LittleEndian.PutUint32(result[frameSize:], payloadSize-4) + copy(result[frameSize+4:], f.Payload) + } + + return result +} + +func buildContextFrame(requestFrame *RntbdFrame) []byte { + builder := RntbdResponseFrameBuilder{} + builder.SetStatusCode(200) + builder.SetResourceType(RntbdResourceTypeConnection) + builder.SetActivityId(requestFrame.ActivityId) + builder.AddHeader(uint16(RntbdContextHeaderServerAgent), RntbdTokenTypeSmallString, "DocumentDB Server") + builder.AddHeader(uint16(RntbdContextHeaderServerVersion), RntbdTokenTypeSmallString, " version=2.14.0.0") + builder.AddHeader(uint16(RntbdContextHeaderIdleTimeoutInSeconds), RntbdTokenTypeULong, uint32(120)) + builder.AddHeader(uint16(RntbdContextHeaderUnauthenticatedTimeoutInSeconds), RntbdTokenTypeULong, uint32(25)) + return builder.Build().ToBytes() +} diff --git a/internal/rntbd/constants.go b/internal/rntbd/constants.go new file mode 100644 index 0000000..c4b757a --- /dev/null +++ b/internal/rntbd/constants.go @@ -0,0 +1,746 @@ +package rntbd + +import ( + "fmt" + + "github.com/pikami/cosmium/api/headers" +) + +type RntbdOperationType uint16 + +const ( + RntbdOperationTypeConnection RntbdOperationType = 0x0000 + RntbdOperationTypeCreate RntbdOperationType = 0x0001 + RntbdOperationTypeUpdate RntbdOperationType = 0x0002 + RntbdOperationTypeRead RntbdOperationType = 0x0003 + RntbdOperationTypeReadFeed RntbdOperationType = 0x0004 + RntbdOperationTypeDelete RntbdOperationType = 0x0005 + RntbdOperationTypeReplace RntbdOperationType = 0x0006 + RntbdOperationTypeExecuteJavaScript RntbdOperationType = 0x0008 + RntbdOperationTypeSQLQuery RntbdOperationType = 0x0009 + RntbdOperationTypePause RntbdOperationType = 0x000A + RntbdOperationTypeResume RntbdOperationType = 0x000B + RntbdOperationTypeStop RntbdOperationType = 0x000C + RntbdOperationTypeRecycle RntbdOperationType = 0x000D + RntbdOperationTypeCrash RntbdOperationType = 0x000E + RntbdOperationTypeQuery RntbdOperationType = 0x000F + RntbdOperationTypeForceConfigRefresh RntbdOperationType = 0x0010 + RntbdOperationTypeHead RntbdOperationType = 0x0011 + RntbdOperationTypeHeadFeed RntbdOperationType = 0x0012 + RntbdOperationTypeUpsert RntbdOperationType = 0x0013 + RntbdOperationTypeRecreate RntbdOperationType = 0x0014 + RntbdOperationTypeThrottle RntbdOperationType = 0x0015 + RntbdOperationTypeGetSplitPoint RntbdOperationType = 0x0016 + RntbdOperationTypePreCreateValidation RntbdOperationType = 0x0017 + RntbdOperationTypeBatchApply RntbdOperationType = 0x0018 + RntbdOperationTypeAbortSplit RntbdOperationType = 0x0019 + RntbdOperationTypeCompleteSplit RntbdOperationType = 0x001A + RntbdOperationTypeOfferUpdateOperation RntbdOperationType = 0x001B + RntbdOperationTypeOfferPreGrowValidation RntbdOperationType = 0x001C + RntbdOperationTypeBatchReportThroughputUtilization RntbdOperationType = 0x001D + RntbdOperationTypeCompletePartitionMigration RntbdOperationType = 0x001E + RntbdOperationTypeAbortPartitionMigration RntbdOperationType = 0x001F + RntbdOperationTypePreReplaceValidation RntbdOperationType = 0x0020 + RntbdOperationTypeAddComputeGatewayRequestCharges RntbdOperationType = 0x0021 + RntbdOperationTypeMigratePartition RntbdOperationType = 0x0022 +) + +type RntbdResourceType uint16 + +const ( + RntbdResourceTypeConnection RntbdResourceType = 0x0000 + RntbdResourceTypeDatabase RntbdResourceType = 0x0001 + RntbdResourceTypeCollection RntbdResourceType = 0x0002 + RntbdResourceTypeDocument RntbdResourceType = 0x0003 + RntbdResourceTypeAttachment RntbdResourceType = 0x0004 + RntbdResourceTypeUser RntbdResourceType = 0x0005 + RntbdResourceTypePermission RntbdResourceType = 0x0006 + RntbdResourceTypeStoredProcedure RntbdResourceType = 0x0007 + RntbdResourceTypeConflict RntbdResourceType = 0x0008 + RntbdResourceTypeTrigger RntbdResourceType = 0x0009 + RntbdResourceTypeUserDefinedFunction RntbdResourceType = 0x000A + RntbdResourceTypeModule RntbdResourceType = 0x000B + RntbdResourceTypeReplica RntbdResourceType = 0x000C + RntbdResourceTypeModuleCommand RntbdResourceType = 0x000D + RntbdResourceTypeRecord RntbdResourceType = 0x000E + RntbdResourceTypeOffer RntbdResourceType = 0x000F + RntbdResourceTypePartitionSetInformation RntbdResourceType = 0x0010 + RntbdResourceTypeXPReplicatorAddress RntbdResourceType = 0x0011 + RntbdResourceTypeMasterPartition RntbdResourceType = 0x0012 + RntbdResourceTypeServerPartition RntbdResourceType = 0x0013 + RntbdResourceTypeDatabaseAccount RntbdResourceType = 0x0014 + RntbdResourceTypeTopology RntbdResourceType = 0x0015 + RntbdResourceTypePartitionKeyRange RntbdResourceType = 0x0016 + RntbdResourceTypeSchema RntbdResourceType = 0x0018 + RntbdResourceTypeBatchApply RntbdResourceType = 0x0019 + RntbdResourceTypeRestoreMetadata RntbdResourceType = 0x001A + RntbdResourceTypeComputeGatewayCharges RntbdResourceType = 0x001B + RntbdResourceTypeRidRange RntbdResourceType = 0x001C + RntbdResourceTypeUserDefinedType RntbdResourceType = 0x001D +) + +type RntbdRequestHeader uint16 + +const ( + RntbdRequestHeaderResourceId RntbdRequestHeader = 0x0000 // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderAuthorizationToken RntbdRequestHeader = 0x0001 // RntbdTokenType.String, required = false + RntbdRequestHeaderPayloadPresent RntbdRequestHeader = 0x0002 // RntbdTokenType.Byte, required = true + RntbdRequestHeaderDate RntbdRequestHeader = 0x0003 // RntbdTokenType.SmallString, required = false + RntbdRequestHeaderPageSize RntbdRequestHeader = 0x0004 // RntbdTokenType.ULong, required = false + RntbdRequestHeaderSessionToken RntbdRequestHeader = 0x0005 // RntbdTokenType.String, required = false + RntbdRequestHeaderContinuationToken RntbdRequestHeader = 0x0006 // RntbdTokenType.String, required = false + RntbdRequestHeaderIndexingDirective RntbdRequestHeader = 0x0007 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderMatch RntbdRequestHeader = 0x0008 // RntbdTokenType.String, required = false + RntbdRequestHeaderPreTriggerInclude RntbdRequestHeader = 0x0009 // RntbdTokenType.String, required = false + RntbdRequestHeaderPostTriggerInclude RntbdRequestHeader = 0x000A // RntbdTokenType.String, required = false + RntbdRequestHeaderIsFanout RntbdRequestHeader = 0x000B // RntbdTokenType.Byte, required = false + RntbdRequestHeaderCollectionPartitionIndex RntbdRequestHeader = 0x000C // RntbdTokenType.ULong, required = false + RntbdRequestHeaderCollectionServiceIndex RntbdRequestHeader = 0x000D // RntbdTokenType.ULong, required = false + RntbdRequestHeaderPreTriggerExclude RntbdRequestHeader = 0x000E // RntbdTokenType.String, required = false + RntbdRequestHeaderPostTriggerExclude RntbdRequestHeader = 0x000F // RntbdTokenType.String, required = false + RntbdRequestHeaderConsistencyLevel RntbdRequestHeader = 0x0010 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderEntityId RntbdRequestHeader = 0x0011 // RntbdTokenType.String, required = false + RntbdRequestHeaderResourceSchemaName RntbdRequestHeader = 0x0012 // RntbdTokenType.SmallString, required = false + RntbdRequestHeaderReplicaPath RntbdRequestHeader = 0x0013 // RntbdTokenType.String, required = true + RntbdRequestHeaderResourceTokenExpiry RntbdRequestHeader = 0x0014 // RntbdTokenType.ULong, required = false + RntbdRequestHeaderDatabaseName RntbdRequestHeader = 0x0015 // RntbdTokenType.String, required = false + RntbdRequestHeaderCollectionName RntbdRequestHeader = 0x0016 // RntbdTokenType.String, required = false + RntbdRequestHeaderDocumentName RntbdRequestHeader = 0x0017 // RntbdTokenType.String, required = false + RntbdRequestHeaderAttachmentName RntbdRequestHeader = 0x0018 // RntbdTokenType.String, required = false + RntbdRequestHeaderUserName RntbdRequestHeader = 0x0019 // RntbdTokenType.String, required = false + RntbdRequestHeaderPermissionName RntbdRequestHeader = 0x001A // RntbdTokenType.String, required = false + RntbdRequestHeaderStoredProcedureName RntbdRequestHeader = 0x001B // RntbdTokenType.String, required = false + RntbdRequestHeaderUserDefinedFunctionName RntbdRequestHeader = 0x001C // RntbdTokenType.String, required = false + RntbdRequestHeaderTriggerName RntbdRequestHeader = 0x001D // RntbdTokenType.String, required = false + RntbdRequestHeaderEnableScanInQuery RntbdRequestHeader = 0x001E // RntbdTokenType.Byte, required = false + RntbdRequestHeaderEmitVerboseTracesInQuery RntbdRequestHeader = 0x001F // RntbdTokenType.Byte, required = false + RntbdRequestHeaderConflictName RntbdRequestHeader = 0x0020 // RntbdTokenType.String, required = false + RntbdRequestHeaderBindReplicaDirective RntbdRequestHeader = 0x0021 // RntbdTokenType.String, required = false + RntbdRequestHeaderPrimaryMasterKey RntbdRequestHeader = 0x0022 // RntbdTokenType.String, required = false + RntbdRequestHeaderSecondaryMasterKey RntbdRequestHeader = 0x0023 // RntbdTokenType.String, required = false + RntbdRequestHeaderPrimaryReadonlyKey RntbdRequestHeader = 0x0024 // RntbdTokenType.String, required = false + RntbdRequestHeaderSecondaryReadonlyKey RntbdRequestHeader = 0x0025 // RntbdTokenType.String, required = false + RntbdRequestHeaderProfileRequest RntbdRequestHeader = 0x0026 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderEnableLowPrecisionOrderBy RntbdRequestHeader = 0x0027 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderClientVersion RntbdRequestHeader = 0x0028 // RntbdTokenType.SmallString, required = false + RntbdRequestHeaderCanCharge RntbdRequestHeader = 0x0029 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderCanThrottle RntbdRequestHeader = 0x002A // RntbdTokenType.Byte, required = false + RntbdRequestHeaderPartitionKey RntbdRequestHeader = 0x002B // RntbdTokenType.String, required = false + RntbdRequestHeaderPartitionKeyRangeId RntbdRequestHeader = 0x002C // RntbdTokenType.String, required = false + RntbdRequestHeaderNotUsed2D RntbdRequestHeader = 0x002D // RntbdTokenType.Invalid, required = false + RntbdRequestHeaderNotUsed2E RntbdRequestHeader = 0x002E // RntbdTokenType.Invalid, required = false + RntbdRequestHeaderNotUsed2F RntbdRequestHeader = 0x002F // RntbdTokenType.Invalid, required = false + RntbdRequestHeaderMigrateCollectionDirective RntbdRequestHeader = 0x0031 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderNotUsed32 RntbdRequestHeader = 0x0032 // RntbdTokenType.Invalid, required = false + RntbdRequestHeaderSupportSpatialLegacyCoordinates RntbdRequestHeader = 0x0033 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderPartitionCount RntbdRequestHeader = 0x0034 // RntbdTokenType.ULong, required = false + RntbdRequestHeaderCollectionRid RntbdRequestHeader = 0x0035 // RntbdTokenType.String, required = false + RntbdRequestHeaderPartitionKeyRangeName RntbdRequestHeader = 0x0036 // RntbdTokenType.String, required = false + RntbdRequestHeaderSchemaName RntbdRequestHeader = 0x003A // RntbdTokenType.String, required = false + RntbdRequestHeaderFilterBySchemaRid RntbdRequestHeader = 0x003B // RntbdTokenType.String, required = false + RntbdRequestHeaderUsePolygonsSmallerThanAHemisphere RntbdRequestHeader = 0x003C // RntbdTokenType.Byte, required = false + RntbdRequestHeaderGatewaySignature RntbdRequestHeader = 0x003D // RntbdTokenType.String, required = false + RntbdRequestHeaderEnableLogging RntbdRequestHeader = 0x003E // RntbdTokenType.Byte, required = false + RntbdRequestHeaderAIM RntbdRequestHeader = 0x003F // RntbdTokenType.String, required = false + RntbdRequestHeaderPopulateQuotaInfo RntbdRequestHeader = 0x0040 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderDisableRUPerMinuteUsage RntbdRequestHeader = 0x0041 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderPopulateQueryMetrics RntbdRequestHeader = 0x0042 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderResponseContinuationTokenLimitInKb RntbdRequestHeader = 0x0043 // RntbdTokenType.ULong, required = false + RntbdRequestHeaderPopulatePartitionStatistics RntbdRequestHeader = 0x0044 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderRemoteStorageType RntbdRequestHeader = 0x0045 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderCollectionRemoteStorageSecurityIdentifier RntbdRequestHeader = 0x0046 // RntbdTokenType.String, required = false + RntbdRequestHeaderIfModifiedSince RntbdRequestHeader = 0x0047 // RntbdTokenType.String, required = false + RntbdRequestHeaderPopulateCollectionThroughputInfo RntbdRequestHeader = 0x0048 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderRemainingTimeInMsOnClientRequest RntbdRequestHeader = 0x0049 // RntbdTokenType.ULong, required = false + RntbdRequestHeaderClientRetryAttemptCount RntbdRequestHeader = 0x004A // RntbdTokenType.ULong, required = false + RntbdRequestHeaderTargetLsn RntbdRequestHeader = 0x004B // RntbdTokenType.LongLong, required = false + RntbdRequestHeaderTargetGlobalCommittedLsn RntbdRequestHeader = 0x004C // RntbdTokenType.LongLong, required = false + RntbdRequestHeaderTransportRequestID RntbdRequestHeader = 0x004D // RntbdTokenType.ULong, required = false + RntbdRequestHeaderRestoreMetadaFilter RntbdRequestHeader = 0x004E // RntbdTokenType.String, required = false + RntbdRequestHeaderRestoreParams RntbdRequestHeader = 0x004F // RntbdTokenType.String, required = false + RntbdRequestHeaderShareThroughput RntbdRequestHeader = 0x0050 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderPartitionResourceFilter RntbdRequestHeader = 0x0051 // RntbdTokenType.String, required = false + RntbdRequestHeaderIsReadOnlyScript RntbdRequestHeader = 0x0052 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderIsAutoScaleRequest RntbdRequestHeader = 0x0053 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderForceQueryScan RntbdRequestHeader = 0x0054 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderCanOfferReplaceComplete RntbdRequestHeader = 0x0056 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderExcludeSystemProperties RntbdRequestHeader = 0x0057 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderBinaryId RntbdRequestHeader = 0x0058 // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderTimeToLiveInSeconds RntbdRequestHeader = 0x0059 // RntbdTokenType.Long, required = false + RntbdRequestHeaderEffectivePartitionKey RntbdRequestHeader = 0x005A // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderBinaryPassthroughRequest RntbdRequestHeader = 0x005B // RntbdTokenType.Byte, required = false + RntbdRequestHeaderUserDefinedTypeName RntbdRequestHeader = 0x005C // RntbdTokenType.String, required = false + RntbdRequestHeaderEnableDynamicRidRangeAllocation RntbdRequestHeader = 0x005D // RntbdTokenType.Byte, required = false + RntbdRequestHeaderEnumerationDirection RntbdRequestHeader = 0x005E // RntbdTokenType.Byte, required = false + RntbdRequestHeaderStartId RntbdRequestHeader = 0x005F // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderEndId RntbdRequestHeader = 0x0060 // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderFanoutOperationState RntbdRequestHeader = 0x0061 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderStartEpk RntbdRequestHeader = 0x0062 // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderEndEpk RntbdRequestHeader = 0x0063 // RntbdTokenType.Bytes, required = false + RntbdRequestHeaderReadFeedKeyType RntbdRequestHeader = 0x0064 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderContentSerializationFormat RntbdRequestHeader = 0x0065 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderAllowTentativeWrites RntbdRequestHeader = 0x0066 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderIsUserRequest RntbdRequestHeader = 0x0067 // RntbdTokenType.Byte, required = false + RntbdRequestHeaderSharedOfferThroughput RntbdRequestHeader = 0x0068 // RntbdTokenType.ULong, required = false + + RntbdRequestHeaderSDKSupportedCapabilities RntbdRequestHeader = 0x00A2 // RntbdTokenType.ULong, required = ? +) + +type RntbdResponseHeaderType uint16 + +const ( + RntbdResponseHeaderPayloadPresent RntbdResponseHeaderType = 0x0000 // RntbdTokenType.Byte, required = true + RntbdResponseHeaderLastStateChangeDateTime RntbdResponseHeaderType = 0x0002 // RntbdTokenType.SmallString, required = false + RntbdResponseHeaderContinuationToken RntbdResponseHeaderType = 0x0003 // RntbdTokenType.String, required = false + RntbdResponseHeaderETag RntbdResponseHeaderType = 0x0004 // RntbdTokenType.String, required = false + RntbdResponseHeaderReadsPerformed RntbdResponseHeaderType = 0x0007 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderWritesPerformed RntbdResponseHeaderType = 0x0008 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderQueriesPerformed RntbdResponseHeaderType = 0x0009 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderIndexTermsGenerated RntbdResponseHeaderType = 0x000A // RntbdTokenType.ULong, required = false + RntbdResponseHeaderScriptsExecuted RntbdResponseHeaderType = 0x000B // RntbdTokenType.ULong, required = false + RntbdResponseHeaderRetryAfterMilliseconds RntbdResponseHeaderType = 0x000C // RntbdTokenType.ULong, required = false + RntbdResponseHeaderIndexingDirective RntbdResponseHeaderType = 0x000D // RntbdTokenType.Byte, required = false + RntbdResponseHeaderStorageMaxResoureQuota RntbdResponseHeaderType = 0x000E // RntbdTokenType.String, required = false + RntbdResponseHeaderStorageResourceQuotaUsage RntbdResponseHeaderType = 0x000F // RntbdTokenType.String, required = false + RntbdResponseHeaderSchemaVersion RntbdResponseHeaderType = 0x0010 // RntbdTokenType.SmallString, required = false + RntbdResponseHeaderCollectionPartitionIndex RntbdResponseHeaderType = 0x0011 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderCollectionServiceIndex RntbdResponseHeaderType = 0x0012 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderLSN RntbdResponseHeaderType = 0x0013 // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderItemCount RntbdResponseHeaderType = 0x0014 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderRequestCharge RntbdResponseHeaderType = 0x0015 // RntbdTokenType.Double, required = false + RntbdResponseHeaderOwnerFullName RntbdResponseHeaderType = 0x0017 // RntbdTokenType.String, required = false + RntbdResponseHeaderOwnerId RntbdResponseHeaderType = 0x0018 // RntbdTokenType.String, required = false + RntbdResponseHeaderDatabaseAccountId RntbdResponseHeaderType = 0x0019 // RntbdTokenType.String, required = false + RntbdResponseHeaderQuorumAckedLSN RntbdResponseHeaderType = 0x001A // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderRequestValidationFailure RntbdResponseHeaderType = 0x001B // RntbdTokenType.Byte, required = false + RntbdResponseHeaderSubStatus RntbdResponseHeaderType = 0x001C // RntbdTokenType.ULong, required = false + RntbdResponseHeaderCollectionUpdateProgress RntbdResponseHeaderType = 0x001D // RntbdTokenType.ULong, required = false + RntbdResponseHeaderCurrentWriteQuorum RntbdResponseHeaderType = 0x001E // RntbdTokenType.ULong, required = false + RntbdResponseHeaderCurrentReplicaSetSize RntbdResponseHeaderType = 0x001F // RntbdTokenType.ULong, required = false + RntbdResponseHeaderCollectionLazyIndexProgress RntbdResponseHeaderType = 0x0020 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderPartitionKeyRangeId RntbdResponseHeaderType = 0x0021 // RntbdTokenType.String, required = false + RntbdResponseHeaderLogResults RntbdResponseHeaderType = 0x0025 // RntbdTokenType.String, required = false + RntbdResponseHeaderXPRole RntbdResponseHeaderType = 0x0026 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderIsRUPerMinuteUsed RntbdResponseHeaderType = 0x0027 // RntbdTokenType.Byte, required = false + RntbdResponseHeaderQueryMetrics RntbdResponseHeaderType = 0x0028 // RntbdTokenType.String, required = false + RntbdResponseHeaderGlobalCommittedLSN RntbdResponseHeaderType = 0x0029 // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderNumberOfReadRegions RntbdResponseHeaderType = 0x0030 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderOfferReplacePending RntbdResponseHeaderType = 0x0031 // RntbdTokenType.Byte, required = false + RntbdResponseHeaderItemLSN RntbdResponseHeaderType = 0x0032 // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderRestoreState RntbdResponseHeaderType = 0x0033 // RntbdTokenType.String, required = false + RntbdResponseHeaderCollectionSecurityIdentifier RntbdResponseHeaderType = 0x0034 // RntbdTokenType.String, required = false + RntbdResponseHeaderTransportRequestID RntbdResponseHeaderType = 0x0035 // RntbdTokenType.ULong, required = false + RntbdResponseHeaderShareThroughput RntbdResponseHeaderType = 0x0036 // RntbdTokenType.Byte, required = false + RntbdResponseHeaderDisableRntbdChannel RntbdResponseHeaderType = 0x0038 // RntbdTokenType.Byte, required = false + RntbdResponseHeaderServerDateTimeUtc RntbdResponseHeaderType = 0x0039 // RntbdTokenType.SmallString, required = false + RntbdResponseHeaderLocalLSN RntbdResponseHeaderType = 0x003A // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderQuorumAckedLocalLSN RntbdResponseHeaderType = 0x003B // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderItemLocalLSN RntbdResponseHeaderType = 0x003C // RntbdTokenType.LongLong, required = false + RntbdResponseHeaderHasTentativeWrites RntbdResponseHeaderType = 0x003D // RntbdTokenType.Byte, required = false + RntbdResponseHeaderSessionToken RntbdResponseHeaderType = 0x003E // RntbdTokenType.String, required = false +) + +type RntbdContextHeader uint16 + +const ( + RntbdContextHeaderProtocolVersion RntbdContextHeader = 0x0000 // RntbdTokenType.ULong, required = false + RntbdContextHeaderClientVersion RntbdContextHeader = 0x0001 // RntbdTokenType.SmallString, required = false + RntbdContextHeaderServerAgent RntbdContextHeader = 0x0002 // RntbdTokenType.SmallString, required = true + RntbdContextHeaderServerVersion RntbdContextHeader = 0x0003 // RntbdTokenType.SmallString, required = true + RntbdContextHeaderIdleTimeoutInSeconds RntbdContextHeader = 0x0004 // RntbdTokenType.ULong, required = false + RntbdContextHeaderUnauthenticatedTimeoutInSeconds RntbdContextHeader = 0x0005 // RntbdTokenType.ULong, required = false +) + +type RntbdTokenType uint8 + +const ( + RntbdTokenTypeByte RntbdTokenType = 0x00 // 8bit boolean + RntbdTokenTypeUShort RntbdTokenType = 0x01 // 16bit unsigned integer + RntbdTokenTypeULong RntbdTokenType = 0x02 // 32bit unsigned integer + RntbdTokenTypeLong RntbdTokenType = 0x03 // 32bit signed integer + RntbdTokenTypeULongLong RntbdTokenType = 0x04 // 64bit unsigned integer + RntbdTokenTypeLongLong RntbdTokenType = 0x05 // 64bit signed integer + RntbdTokenTypeGuid RntbdTokenType = 0x06 // 128bit GUID + RntbdTokenTypeSmallString RntbdTokenType = 0x07 // 8bit len + string + RntbdTokenTypeString RntbdTokenType = 0x08 // 16bit len + string + RntbdTokenTypeULongString RntbdTokenType = 0x09 // 32bit len + string + RntbdTokenTypeSmallBytes RntbdTokenType = 0x0A // 8bit len + bytes + RntbdTokenTypeBytes RntbdTokenType = 0x0B // 16bit len + bytes + RntbdTokenTypeULongBytes RntbdTokenType = 0x0C // 32bit len + bytes + RntbdTokenTypeFloat RntbdTokenType = 0x0D // 32bit float + RntbdTokenTypeDouble RntbdTokenType = 0x0E // 64bit double + RntbdTokenTypeInvalid RntbdTokenType = 0x0F // Invalid token type +) + +func (h RntbdRequestHeader) String() string { + switch h { + case RntbdRequestHeaderResourceId: + return "RntbdRequestHeaderResourceId" + case RntbdRequestHeaderAuthorizationToken: + return headers.Authorization + case RntbdRequestHeaderPayloadPresent: + return "RntbdRequestHeaderPayloadPresent" + case RntbdRequestHeaderDate: + return headers.XDate + case RntbdRequestHeaderPageSize: + return "RntbdRequestHeaderPageSize" + case RntbdRequestHeaderSessionToken: + return "RntbdRequestHeaderSessionToken" + case RntbdRequestHeaderContinuationToken: + return "RntbdRequestHeaderContinuationToken" + case RntbdRequestHeaderIndexingDirective: + return "RntbdRequestHeaderIndexingDirective" + case RntbdRequestHeaderMatch: + return "RntbdRequestHeaderMatch" + case RntbdRequestHeaderPreTriggerInclude: + return "RntbdRequestHeaderPreTriggerInclude" + case RntbdRequestHeaderPostTriggerInclude: + return "RntbdRequestHeaderPostTriggerInclude" + case RntbdRequestHeaderIsFanout: + return "RntbdRequestHeaderIsFanout" + case RntbdRequestHeaderCollectionPartitionIndex: + return "RntbdRequestHeaderCollectionPartitionIndex" + case RntbdRequestHeaderCollectionServiceIndex: + return "RntbdRequestHeaderCollectionServiceIndex" + case RntbdRequestHeaderPreTriggerExclude: + return "RntbdRequestHeaderPreTriggerExclude" + case RntbdRequestHeaderPostTriggerExclude: + return "RntbdRequestHeaderPostTriggerExclude" + case RntbdRequestHeaderConsistencyLevel: + return headers.ConsistencyLevel + case RntbdRequestHeaderEntityId: + return "RntbdRequestHeaderEntityId" + case RntbdRequestHeaderResourceSchemaName: + return "RntbdRequestHeaderResourceSchemaName" + case RntbdRequestHeaderReplicaPath: + return "RntbdRequestHeaderReplicaPath" + case RntbdRequestHeaderResourceTokenExpiry: + return "RntbdRequestHeaderResourceTokenExpiry" + case RntbdRequestHeaderDatabaseName: + return "RntbdRequestHeaderDatabaseName" + case RntbdRequestHeaderCollectionName: + return "RntbdRequestHeaderCollectionName" + case RntbdRequestHeaderDocumentName: + return "RntbdRequestHeaderDocumentName" + case RntbdRequestHeaderAttachmentName: + return "RntbdRequestHeaderAttachmentName" + case RntbdRequestHeaderUserName: + return "RntbdRequestHeaderUserName" + case RntbdRequestHeaderPermissionName: + return "RntbdRequestHeaderPermissionName" + case RntbdRequestHeaderStoredProcedureName: + return "RntbdRequestHeaderStoredProcedureName" + case RntbdRequestHeaderUserDefinedFunctionName: + return "RntbdRequestHeaderUserDefinedFunctionName" + case RntbdRequestHeaderTriggerName: + return "RntbdRequestHeaderTriggerName" + case RntbdRequestHeaderEnableScanInQuery: + return "RntbdRequestHeaderEnableScanInQuery" + case RntbdRequestHeaderEmitVerboseTracesInQuery: + return "RntbdRequestHeaderEmitVerboseTracesInQuery" + case RntbdRequestHeaderConflictName: + return "RntbdRequestHeaderConflictName" + case RntbdRequestHeaderBindReplicaDirective: + return "RntbdRequestHeaderBindReplicaDirective" + case RntbdRequestHeaderPrimaryMasterKey: + return "RntbdRequestHeaderPrimaryMasterKey" + case RntbdRequestHeaderSecondaryMasterKey: + return "RntbdRequestHeaderSecondaryMasterKey" + case RntbdRequestHeaderPrimaryReadonlyKey: + return "RntbdRequestHeaderPrimaryReadonlyKey" + case RntbdRequestHeaderSecondaryReadonlyKey: + return "RntbdRequestHeaderSecondaryReadonlyKey" + case RntbdRequestHeaderProfileRequest: + return "RntbdRequestHeaderProfileRequest" + case RntbdRequestHeaderEnableLowPrecisionOrderBy: + return "RntbdRequestHeaderEnableLowPrecisionOrderBy" + case RntbdRequestHeaderClientVersion: + return "RntbdRequestHeaderClientVersion" + case RntbdRequestHeaderCanCharge: + return "RntbdRequestHeaderCanCharge" + case RntbdRequestHeaderCanThrottle: + return "RntbdRequestHeaderCanThrottle" + case RntbdRequestHeaderPartitionKey: + return "RntbdRequestHeaderPartitionKey" + case RntbdRequestHeaderPartitionKeyRangeId: + return "RntbdRequestHeaderPartitionKeyRangeId" + case RntbdRequestHeaderNotUsed2D: + return "RntbdRequestHeaderNotUsed2D" + case RntbdRequestHeaderNotUsed2E: + return "RntbdRequestHeaderNotUsed2E" + case RntbdRequestHeaderNotUsed2F: + return "RntbdRequestHeaderNotUsed2F" + case RntbdRequestHeaderMigrateCollectionDirective: + return "RntbdRequestHeaderMigrateCollectionDirective" + case RntbdRequestHeaderNotUsed32: + return "RntbdRequestHeaderNotUsed32" + case RntbdRequestHeaderSupportSpatialLegacyCoordinates: + return "RntbdRequestHeaderSupportSpatialLegacyCoordinates" + case RntbdRequestHeaderPartitionCount: + return "RntbdRequestHeaderPartitionCount" + case RntbdRequestHeaderCollectionRid: + return "RntbdRequestHeaderCollectionRid" + case RntbdRequestHeaderPartitionKeyRangeName: + return "RntbdRequestHeaderPartitionKeyRangeName" + case RntbdRequestHeaderSchemaName: + return "RntbdRequestHeaderSchemaName" + case RntbdRequestHeaderFilterBySchemaRid: + return "RntbdRequestHeaderFilterBySchemaRid" + case RntbdRequestHeaderUsePolygonsSmallerThanAHemisphere: + return "RntbdRequestHeaderUsePolygonsSmallerThanAHemisphere" + case RntbdRequestHeaderGatewaySignature: + return "RntbdRequestHeaderGatewaySignature" + case RntbdRequestHeaderEnableLogging: + return "RntbdRequestHeaderEnableLogging" + case RntbdRequestHeaderAIM: + return headers.AIM + case RntbdRequestHeaderPopulateQuotaInfo: + return "RntbdRequestHeaderPopulateQuotaInfo" + case RntbdRequestHeaderDisableRUPerMinuteUsage: + return "RntbdRequestHeaderDisableRUPerMinuteUsage" + case RntbdRequestHeaderPopulateQueryMetrics: + return "RntbdRequestHeaderPopulateQueryMetrics" + case RntbdRequestHeaderResponseContinuationTokenLimitInKb: + return "RntbdRequestHeaderResponseContinuationTokenLimitInKb" + case RntbdRequestHeaderPopulatePartitionStatistics: + return "RntbdRequestHeaderPopulatePartitionStatistics" + case RntbdRequestHeaderRemoteStorageType: + return "RntbdRequestHeaderRemoteStorageType" + case RntbdRequestHeaderCollectionRemoteStorageSecurityIdentifier: + return "RntbdRequestHeaderCollectionRemoteStorageSecurityIdentifier" + case RntbdRequestHeaderIfModifiedSince: + return "RntbdRequestHeaderIfModifiedSince" + case RntbdRequestHeaderPopulateCollectionThroughputInfo: + return "RntbdRequestHeaderPopulateCollectionThroughputInfo" + case RntbdRequestHeaderRemainingTimeInMsOnClientRequest: + return headers.RemainingTimeInMsOnClient + case RntbdRequestHeaderClientRetryAttemptCount: + return headers.ClientRetryAttemptCount + case RntbdRequestHeaderTargetLsn: + return "RntbdRequestHeaderTargetLsn" + case RntbdRequestHeaderTargetGlobalCommittedLsn: + return "RntbdRequestHeaderTargetGlobalCommittedLsn" + case RntbdRequestHeaderTransportRequestID: + return "RntbdRequestHeaderTransportRequestID" + case RntbdRequestHeaderRestoreMetadaFilter: + return "RntbdRequestHeaderRestoreMetadaFilter" + case RntbdRequestHeaderRestoreParams: + return "RntbdRequestHeaderRestoreParams" + case RntbdRequestHeaderShareThroughput: + return "RntbdRequestHeaderShareThroughput" + case RntbdRequestHeaderPartitionResourceFilter: + return "RntbdRequestHeaderPartitionResourceFilter" + case RntbdRequestHeaderIsReadOnlyScript: + return "RntbdRequestHeaderIsReadOnlyScript" + case RntbdRequestHeaderIsAutoScaleRequest: + return "RntbdRequestHeaderIsAutoScaleRequest" + case RntbdRequestHeaderForceQueryScan: + return "RntbdRequestHeaderForceQueryScan" + case RntbdRequestHeaderCanOfferReplaceComplete: + return "RntbdRequestHeaderCanOfferReplaceComplete" + case RntbdRequestHeaderExcludeSystemProperties: + return "RntbdRequestHeaderExcludeSystemProperties" + case RntbdRequestHeaderBinaryId: + return "RntbdRequestHeaderBinaryId" + case RntbdRequestHeaderTimeToLiveInSeconds: + return "RntbdRequestHeaderTimeToLiveInSeconds" + case RntbdRequestHeaderEffectivePartitionKey: + return "RntbdRequestHeaderEffectivePartitionKey" + case RntbdRequestHeaderBinaryPassthroughRequest: + return "RntbdRequestHeaderBinaryPassthroughRequest" + case RntbdRequestHeaderUserDefinedTypeName: + return "RntbdRequestHeaderUserDefinedTypeName" + case RntbdRequestHeaderEnableDynamicRidRangeAllocation: + return "RntbdRequestHeaderEnableDynamicRidRangeAllocation" + case RntbdRequestHeaderEnumerationDirection: + return "RntbdRequestHeaderEnumerationDirection" + case RntbdRequestHeaderStartId: + return "RntbdRequestHeaderStartId" + case RntbdRequestHeaderEndId: + return "RntbdRequestHeaderEndId" + case RntbdRequestHeaderFanoutOperationState: + return "RntbdRequestHeaderFanoutOperationState" + case RntbdRequestHeaderStartEpk: + return "RntbdRequestHeaderStartEpk" + case RntbdRequestHeaderEndEpk: + return "RntbdRequestHeaderEndEpk" + case RntbdRequestHeaderReadFeedKeyType: + return "RntbdRequestHeaderReadFeedKeyType" + case RntbdRequestHeaderContentSerializationFormat: + return "RntbdRequestHeaderContentSerializationFormat" + case RntbdRequestHeaderAllowTentativeWrites: + return "RntbdRequestHeaderAllowTentativeWrites" + case RntbdRequestHeaderIsUserRequest: + return "RntbdRequestHeaderIsUserRequest" + case RntbdRequestHeaderSharedOfferThroughput: + return "RntbdRequestHeaderSharedOfferThroughput" + case RntbdRequestHeaderSDKSupportedCapabilities: + return headers.SupportedCapabilities + } + + return fmt.Sprintf("RntbdRequestHeader(%d)", h) +} + +func (h RntbdContextHeader) String() string { + switch h { + case RntbdContextHeaderProtocolVersion: + return "RntbdContextHeaderProtocolVersion" + case RntbdContextHeaderClientVersion: + return "RntbdContextHeaderClientVersion" + case RntbdContextHeaderServerAgent: + return "RntbdContextHeaderServerAgent" + case RntbdContextHeaderServerVersion: + return "RntbdContextHeaderServerVersion" + case RntbdContextHeaderIdleTimeoutInSeconds: + return "RntbdContextHeaderIdleTimeoutInSeconds" + case RntbdContextHeaderUnauthenticatedTimeoutInSeconds: + return "RntbdContextHeaderUnauthenticatedTimeoutInSeconds" + } + + return fmt.Sprintf("RntbdContextHeader(%d)", h) +} + +func (h RntbdResponseHeaderType) String() string { + switch h { + case RntbdResponseHeaderPayloadPresent: + return "PayloadPresent" + case RntbdResponseHeaderLastStateChangeDateTime: + return "LastStateChangeDateTime" + case RntbdResponseHeaderContinuationToken: + return "ContinuationToken" + case RntbdResponseHeaderETag: + return "ETag" + case RntbdResponseHeaderReadsPerformed: + return "ReadsPerformed" + case RntbdResponseHeaderWritesPerformed: + return "WritesPerformed" + case RntbdResponseHeaderQueriesPerformed: + return "QueriesPerformed" + case RntbdResponseHeaderIndexTermsGenerated: + return "IndexTermsGenerated" + case RntbdResponseHeaderScriptsExecuted: + return "ScriptsExecuted" + case RntbdResponseHeaderRetryAfterMilliseconds: + return "RetryAfterMilliseconds" + case RntbdResponseHeaderIndexingDirective: + return "IndexingDirective" + case RntbdResponseHeaderStorageMaxResoureQuota: + return "StorageMaxResoureQuota" + case RntbdResponseHeaderStorageResourceQuotaUsage: + return "StorageResourceQuotaUsage" + case RntbdResponseHeaderSchemaVersion: + return "SchemaVersion" + case RntbdResponseHeaderCollectionPartitionIndex: + return "CollectionPartitionIndex" + case RntbdResponseHeaderCollectionServiceIndex: + return "CollectionServiceIndex" + case RntbdResponseHeaderLSN: + return "LSN" + case RntbdResponseHeaderItemCount: + return "ItemCount" + case RntbdResponseHeaderRequestCharge: + return "RequestCharge" + case RntbdResponseHeaderOwnerFullName: + return "OwnerFullName" + case RntbdResponseHeaderOwnerId: + return "OwnerId" + case RntbdResponseHeaderDatabaseAccountId: + return "DatabaseAccountId" + case RntbdResponseHeaderQuorumAckedLSN: + return "QuorumAckedLSN" + case RntbdResponseHeaderRequestValidationFailure: + return "RequestValidationFailure" + case RntbdResponseHeaderSubStatus: + return "SubStatus" + case RntbdResponseHeaderCollectionUpdateProgress: + return "CollectionUpdateProgress" + case RntbdResponseHeaderCurrentWriteQuorum: + return "CurrentWriteQuorum" + case RntbdResponseHeaderCurrentReplicaSetSize: + return "CurrentReplicaSetSize" + case RntbdResponseHeaderCollectionLazyIndexProgress: + return "CollectionLazyIndexProgress" + case RntbdResponseHeaderPartitionKeyRangeId: + return "PartitionKeyRangeId" + case RntbdResponseHeaderLogResults: + return "LogResults" + case RntbdResponseHeaderXPRole: + return "XPRole" + case RntbdResponseHeaderIsRUPerMinuteUsed: + return "IsRUPerMinuteUsed" + case RntbdResponseHeaderQueryMetrics: + return "QueryMetrics" + case RntbdResponseHeaderGlobalCommittedLSN: + return "GlobalCommittedLSN" + case RntbdResponseHeaderNumberOfReadRegions: + return "NumberOfReadRegions" + case RntbdResponseHeaderOfferReplacePending: + return "OfferReplacePending" + case RntbdResponseHeaderItemLSN: + return "ItemLSN" + case RntbdResponseHeaderRestoreState: + return "RestoreState" + case RntbdResponseHeaderCollectionSecurityIdentifier: + return "CollectionSecurityIdentifier" + case RntbdResponseHeaderTransportRequestID: + return "TransportRequestID" + case RntbdResponseHeaderShareThroughput: + return "ShareThroughput" + case RntbdResponseHeaderDisableRntbdChannel: + return "DisableRntbdChannel" + case RntbdResponseHeaderServerDateTimeUtc: + return "ServerDateTimeUtc" + case RntbdResponseHeaderLocalLSN: + return "LocalLSN" + case RntbdResponseHeaderQuorumAckedLocalLSN: + return "QuorumAckedLocalLSN" + case RntbdResponseHeaderItemLocalLSN: + return "ItemLocalLSN" + case RntbdResponseHeaderHasTentativeWrites: + return "HasTentativeWrites" + case RntbdResponseHeaderSessionToken: + return "SessionToken" + } + + return fmt.Sprintf("RntbdResponseHeaderType(%d)", h) +} + +func (r RntbdResourceType) String() string { + switch r { + case RntbdResourceTypeConnection: + return "Connection" + case RntbdResourceTypeDatabase: + return "Database" + case RntbdResourceTypeCollection: + return "Collection" + case RntbdResourceTypeDocument: + return "Document" + case RntbdResourceTypeAttachment: + return "Attachment" + case RntbdResourceTypeUser: + return "User" + case RntbdResourceTypePermission: + return "Permission" + case RntbdResourceTypeStoredProcedure: + return "StoredProcedure" + case RntbdResourceTypeConflict: + return "Conflict" + case RntbdResourceTypeTrigger: + return "Trigger" + case RntbdResourceTypeUserDefinedFunction: + return "UserDefinedFunction" + case RntbdResourceTypeModule: + return "Module" + case RntbdResourceTypeReplica: + return "Replica" + case RntbdResourceTypeModuleCommand: + return "ModuleCommand" + case RntbdResourceTypeRecord: + return "Record" + case RntbdResourceTypeOffer: + return "Offer" + case RntbdResourceTypePartitionSetInformation: + return "PartitionSetInformation" + case RntbdResourceTypeXPReplicatorAddress: + return "XPReplicatorAddress" + case RntbdResourceTypeMasterPartition: + return "MasterPartition" + case RntbdResourceTypeServerPartition: + return "ServerPartition" + case RntbdResourceTypeDatabaseAccount: + return "DatabaseAccount" + case RntbdResourceTypeTopology: + return "Topology" + case RntbdResourceTypePartitionKeyRange: + return "PartitionKeyRange" + case RntbdResourceTypeSchema: + return "Schema" + case RntbdResourceTypeBatchApply: + return "BatchApply" + case RntbdResourceTypeRestoreMetadata: + return "RestoreMetadata" + case RntbdResourceTypeComputeGatewayCharges: + return "ComputeGatewayCharges" + case RntbdResourceTypeRidRange: + return "RidRange" + case RntbdResourceTypeUserDefinedType: + return "UserDefinedType" + } + + return fmt.Sprintf("RntbdResourceType(%d)", r) +} + +func (o RntbdOperationType) String() string { + switch o { + case RntbdOperationTypeConnection: + return "Connection" + case RntbdOperationTypeCreate: + return "Create" + case RntbdOperationTypeUpdate: + return "Update" + case RntbdOperationTypeRead: + return "Read" + case RntbdOperationTypeReadFeed: + return "ReadFeed" + case RntbdOperationTypeDelete: + return "Delete" + case RntbdOperationTypeReplace: + return "Replace" + case RntbdOperationTypeExecuteJavaScript: + return "ExecuteJavaScript" + case RntbdOperationTypeSQLQuery: + return "SQLQuery" + case RntbdOperationTypePause: + return "Pause" + case RntbdOperationTypeResume: + return "Resume" + case RntbdOperationTypeStop: + return "Stop" + case RntbdOperationTypeRecycle: + return "Recycle" + case RntbdOperationTypeCrash: + return "Crash" + case RntbdOperationTypeQuery: + return "Query" + case RntbdOperationTypeForceConfigRefresh: + return "ForceConfigRefresh" + case RntbdOperationTypeHead: + return "Head" + case RntbdOperationTypeHeadFeed: + return "HeadFeed" + case RntbdOperationTypeUpsert: + return "Upsert" + case RntbdOperationTypeRecreate: + return "Recreate" + case RntbdOperationTypeThrottle: + return "Throttle" + case RntbdOperationTypeGetSplitPoint: + return "GetSplitPoint" + case RntbdOperationTypePreCreateValidation: + return "PreCreateValidation" + case RntbdOperationTypeBatchApply: + return "BatchApply" + case RntbdOperationTypeAbortSplit: + return "AbortSplit" + case RntbdOperationTypeCompleteSplit: + return "CompleteSplit" + case RntbdOperationTypeOfferUpdateOperation: + return "OfferUpdateOperation" + case RntbdOperationTypeOfferPreGrowValidation: + return "OfferPreGrowValidation" + case RntbdOperationTypeBatchReportThroughputUtilization: + return "BatchReportThroughputUtilization" + case RntbdOperationTypeCompletePartitionMigration: + return "CompletePartitionMigration" + case RntbdOperationTypeAbortPartitionMigration: + return "AbortPartitionMigration" + case RntbdOperationTypePreReplaceValidation: + return "PreReplaceValidation" + case RntbdOperationTypeAddComputeGatewayRequestCharges: + return "AddComputeGatewayRequestCharges" + case RntbdOperationTypeMigratePartition: + return "MigratePartition" + } + + return fmt.Sprintf("RntbdOperationType(%d)", o) +} diff --git a/internal/rntbd/converter.go b/internal/rntbd/converter.go new file mode 100644 index 0000000..f31c33f --- /dev/null +++ b/internal/rntbd/converter.go @@ -0,0 +1,128 @@ +package rntbd + +import ( + "bytes" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + + "github.com/pikami/cosmium/api/headers" +) + +func (f *RntbdFrame) ToHttpRequest() *http.Request { + req := &http.Request{ + Method: operationTypeToHttpMethod(f.OperationType), + URL: &url.URL{Path: frameToPath(f)}, + Body: io.NopCloser(bytes.NewReader(f.Payload)), + Header: http.Header{}, + } + + switch f.OperationType { + case RntbdOperationTypeQuery, RntbdOperationTypeSQLQuery: + req.Header.Set(headers.Query, "true") + case RntbdOperationTypeUpsert: + req.Header.Set(headers.IsUpsert, "true") + } + + if ifMatch, ok := f.RequestHeaders[RntbdRequestHeaderMatch]; ok { + if ifMatchString, ok := ifMatch.(string); ok { + req.Header.Set(headers.IfMatch, ifMatchString) + } + } + + if continuationToken, ok := f.RequestHeaders[RntbdRequestHeaderContinuationToken]; ok { + if continuationTokenString, ok := continuationToken.(string); ok { + req.Header.Set(headers.ContinuationToken, continuationTokenString) + } + } + + if maxItemCount, ok := f.RequestHeaders[RntbdRequestHeaderPageSize]; ok { + if maxItemCountString, ok := maxItemCount.(uint64); ok { + req.Header.Set(headers.MaxItemCount, fmt.Sprintf("%d", maxItemCountString)) + } + } + + return req +} + +func ToRntbdResponseFrame(responseWriter *httptest.ResponseRecorder) *RntbdResponseFrameBuilder { + builder := &RntbdResponseFrameBuilder{} + builder.SetStatusCode(uint16(responseWriter.Code)) + + if responseWriter.Header().Get(headers.ETag) != "" { + builder.AddHeader(uint16(RntbdResponseHeaderETag), RntbdTokenTypeString, responseWriter.Header().Get(headers.ETag)) + } + + if responseWriter.Header().Get(headers.ContinuationToken) != "" { + builder.AddHeader(uint16(RntbdResponseHeaderContinuationToken), RntbdTokenTypeString, responseWriter.Header().Get(headers.ContinuationToken)) + } + + if responseWriter.Header().Get(headers.ItemCount) != "" { + itemCount, err := strconv.ParseUint(responseWriter.Header().Get(headers.ItemCount), 10, 32) + if err != nil { + panic(err) + } + + builder.AddHeader(uint16(RntbdResponseHeaderItemCount), RntbdTokenTypeULong, uint32(itemCount)) + } + + if responseWriter.Body.Len() > 0 { + builder.AddHeader(uint16(RntbdResponseHeaderPayloadPresent), RntbdTokenTypeByte, []byte{1}) + builder.AddPayload(responseWriter.Body.Bytes()) + } else { + builder.AddHeader(uint16(RntbdResponseHeaderPayloadPresent), RntbdTokenTypeByte, []byte{0}) + } + + return builder +} + +func operationTypeToHttpMethod(operationType RntbdOperationType) string { + switch operationType { + case RntbdOperationTypeRead, + RntbdOperationTypeReadFeed: + return http.MethodGet + case RntbdOperationTypeCreate, + RntbdOperationTypeUpsert, + RntbdOperationTypeQuery, + RntbdOperationTypeSQLQuery: + return http.MethodPost + case RntbdOperationTypeUpdate, + RntbdOperationTypeReplace: + return http.MethodPut + case RntbdOperationTypeDelete: + return http.MethodDelete + } + + panic(fmt.Sprintf("Unknown operation type: %d", operationType)) +} + +func frameToPath(frame *RntbdFrame) string { + databaseName, databaseOk := frame.RequestHeaders[RntbdRequestHeaderDatabaseName] + collectionName, collectionOk := frame.RequestHeaders[RntbdRequestHeaderCollectionName] + documentName, documentOk := frame.RequestHeaders[RntbdRequestHeaderDocumentName] + + urlPath := "" + + if databaseOk { + urlPath += fmt.Sprintf("/dbs/%s", databaseName) + } else if frame.ResourceType == RntbdResourceTypeDatabase { + urlPath += "/dbs" + } + + if collectionOk { + urlPath += fmt.Sprintf("/colls/%s", collectionName) + } else if frame.ResourceType == RntbdResourceTypeCollection { + urlPath += "/colls" + } + + if documentOk { + urlPath += fmt.Sprintf("/docs/%s", documentName) + } else if frame.ResourceType == RntbdResourceTypeDocument { + urlPath += "/docs" + } + + return urlPath +} diff --git a/internal/rntbd/parser.go b/internal/rntbd/parser.go new file mode 100644 index 0000000..2c329fd --- /dev/null +++ b/internal/rntbd/parser.go @@ -0,0 +1,212 @@ +package rntbd + +import ( + "bufio" + "bytes" + "encoding/binary" + "encoding/hex" + "fmt" + "io" + "os" + + "github.com/pikami/cosmium/internal/logger" +) + +type RntbdFrame struct { + ResourceType RntbdResourceType + OperationType RntbdOperationType + ActivityId []byte + RequestHeaders map[RntbdRequestHeader]any + ResponseHeaders map[RntbdResponseHeaderType]any + ContextHeaders map[RntbdContextHeader]any + Payload []byte +} + +func ReadFrame(reader *bufio.Reader) (*RntbdFrame, error) { + sizeBytes := readBytes(reader, 4) + size := binary.LittleEndian.Uint32(sizeBytes) + + payload := readBytes(reader, int(size)-4) + + frame, err := parseFrame_Int(payload, false) + if err != nil { + return nil, err + } + + if payloadPresent, ok := frame.RequestHeaders[RntbdRequestHeaderPayloadPresent]; ok && payloadPresent.([]byte)[0] == 1 { + payloadSize := binary.LittleEndian.Uint32(readBytes(reader, 4)) + payload := readBytes(reader, int(payloadSize)) + frame.Payload = payload + } + + if payloadPresent, ok := frame.ResponseHeaders[RntbdResponseHeaderPayloadPresent]; ok && payloadPresent.([]byte)[0] == 1 { + payloadSize := binary.LittleEndian.Uint32(readBytes(reader, 4)) + payload := readBytes(reader, int(payloadSize)) + frame.Payload = payload + } + + return frame, nil +} + +func ParseFrame(data []byte, isResponse bool) (*RntbdFrame, error) { + if len(data) < 4 { + return nil, fmt.Errorf("data too short") + } + + reader := bufio.NewReader(bytes.NewReader(data)) + sizeBytes := readBytes(reader, 4) + size := binary.LittleEndian.Uint32(sizeBytes) + + payload := readBytes(reader, int(size)-4) + frame, err := parseFrame_Int(payload, isResponse) + if err != nil { + return nil, err + } + + if payloadPresent, ok := frame.RequestHeaders[RntbdRequestHeaderPayloadPresent]; ok && payloadPresent.([]byte)[0] == 1 { + payloadSize := binary.LittleEndian.Uint32(readBytes(reader, 4)) + payload := readBytes(reader, int(payloadSize)) + frame.Payload = payload + } + + if payloadPresent, ok := frame.ResponseHeaders[RntbdResponseHeaderPayloadPresent]; ok && payloadPresent.([]byte)[0] == 1 { + payloadSize := binary.LittleEndian.Uint32(readBytes(reader, 4)) + payload := readBytes(reader, int(payloadSize)) + frame.Payload = payload + } + + leftOverBytes, err := io.ReadAll(reader) + if err != nil { + logger.ErrorLn("Error reading leftOverBytes:", err) + } + + if len(leftOverBytes) > 0 { + logger.ErrorLn("Left over bytes:", hex.EncodeToString(leftOverBytes)) + } + + return frame, nil +} + +func parseFrame_Int(data []byte, isResponse bool) (*RntbdFrame, error) { + payloadReader := bufio.NewReader(bytes.NewReader(data)) + + resourceTypeBytes := readBytes(payloadReader, 2) + resourceType := binary.LittleEndian.Uint16(resourceTypeBytes) + + operationTypeBytes := readBytes(payloadReader, 2) + operationType := RntbdOperationType(binary.LittleEndian.Uint16(operationTypeBytes)) + + activityIdBytes := readBytes(payloadReader, 16) + + requestHeaders := make(map[RntbdRequestHeader]any) + responseHeaders := make(map[RntbdResponseHeaderType]any) + contextHeaders := make(map[RntbdContextHeader]any) + for { + if _, err := payloadReader.Peek(1); err != nil { + break + } + + headerIdBytes := readBytes(payloadReader, 2) + headerId := binary.LittleEndian.Uint16(headerIdBytes) + + token, err := parseRntbdToken(payloadReader) + if err != nil { + return nil, err + } + + if resourceType == uint16(RntbdResourceTypeConnection) { + contextHeaders[RntbdContextHeader(headerId)] = token + } else if isResponse { + responseHeaders[RntbdResponseHeaderType(headerId)] = token + } else { + requestHeaders[RntbdRequestHeader(headerId)] = token + } + } + + return &RntbdFrame{ + ResourceType: RntbdResourceType(resourceType), + OperationType: RntbdOperationType(operationType), + ActivityId: activityIdBytes, + RequestHeaders: requestHeaders, + ResponseHeaders: responseHeaders, + ContextHeaders: contextHeaders, + }, nil +} + +func parseRntbdToken(reader *bufio.Reader) (any, error) { + tokenTypeBytes := readBytes(reader, 1) + tokenType := RntbdTokenType(tokenTypeBytes[0]) + + switch tokenType { + case RntbdTokenTypeByte: + token := readBytes(reader, 1) + return token, nil + case RntbdTokenTypeUShort: + token := binary.LittleEndian.Uint16(readBytes(reader, 2)) + return token, nil + case RntbdTokenTypeULong: + token := binary.LittleEndian.Uint32(readBytes(reader, 4)) + return token, nil + case RntbdTokenTypeLong: + token := int32(binary.LittleEndian.Uint32(readBytes(reader, 4))) + return token, nil + case RntbdTokenTypeULongLong: + token := binary.LittleEndian.Uint64(readBytes(reader, 8)) + return token, nil + case RntbdTokenTypeLongLong: + token := int64(binary.LittleEndian.Uint64(readBytes(reader, 8))) + return token, nil + case RntbdTokenTypeGuid: + token := readBytes(reader, 16) + return token, nil + case RntbdTokenTypeSmallString: + lengthBytes := readBytes(reader, 1) + length := uint8(lengthBytes[0]) + token := readBytes(reader, int(length)) + return string(token), nil + case RntbdTokenTypeString: + length := binary.LittleEndian.Uint16(readBytes(reader, 2)) + token := readBytes(reader, int(length)) + return string(token), nil + case RntbdTokenTypeULongString: + length := binary.LittleEndian.Uint32(readBytes(reader, 4)) + token := readBytes(reader, int(length)) + return string(token), nil + case RntbdTokenTypeSmallBytes: + lengthBytes := readBytes(reader, 1) + length := uint8(lengthBytes[0]) + token := readBytes(reader, int(length)) + return token, nil + case RntbdTokenTypeBytes: + length := binary.LittleEndian.Uint16(readBytes(reader, 2)) + token := readBytes(reader, int(length)) + return token, nil + case RntbdTokenTypeULongBytes: + length := binary.LittleEndian.Uint32(readBytes(reader, 4)) + token := readBytes(reader, int(length)) + return token, nil + case RntbdTokenTypeFloat: + // I can't be bothered to implement this, let's just return a byte array + token := readBytes(reader, 4) + return token, nil + case RntbdTokenTypeDouble: + // I can't be bothered to implement this, let's just return a byte array + token := readBytes(reader, 8) + return token, nil + case RntbdTokenTypeInvalid: + return nil, fmt.Errorf("invalid token type") + } + + return nil, fmt.Errorf("invalid token type") +} + +func readBytes(reader *bufio.Reader, n int) []byte { + bytes := make([]byte, n) + _, err := io.ReadFull(reader, bytes) + if err != nil { + logger.ErrorLn("Error reading bytes:", err) + os.Exit(0) + } + + return bytes +} diff --git a/internal/rntbd/server.go b/internal/rntbd/server.go new file mode 100644 index 0000000..4e14249 --- /dev/null +++ b/internal/rntbd/server.go @@ -0,0 +1,120 @@ +package rntbd + +import ( + "bufio" + "crypto/tls" + "fmt" + "net" + "net/http/httptest" + + "github.com/pikami/cosmium/api" + "github.com/pikami/cosmium/internal/logger" + tlsprovider "github.com/pikami/cosmium/internal/tls_provider" +) + +type RntbdServer struct { + port int + listener net.Listener + apiServer *api.ApiServer +} + +func NewRntbdServer(port int, apiServer *api.ApiServer) *RntbdServer { + return &RntbdServer{port: port, apiServer: apiServer} +} + +func (s *RntbdServer) Start() error { + tlsConfig := tlsprovider.GetDefaultTlsConfig() + listener, err := tls.Listen("tcp", fmt.Sprintf(":%d", s.port), tlsConfig) + if err != nil { + return fmt.Errorf("failed to listen on port %d: %w", s.port, err) + } + s.listener = listener + + go func() { + for { + conn, err := s.listener.Accept() + if err != nil { + logger.ErrorLn("Failed to accept connection:", err) + continue + } + + go s.handleConnection(conn) + } + }() + + return nil +} + +func (s *RntbdServer) Stop() error { + return s.listener.Close() +} + +func (s *RntbdServer) handleConnection(conn net.Conn) { + defer conn.Close() + + reader := bufio.NewReader(conn) + writer := bufio.NewWriter(conn) + + for { + _, err := reader.Peek(4) + if err != nil { + return + } + + frame, err := ReadFrame(reader) + if err != nil { + logger.ErrorLn("Failed to read frame:", err) + continue + } + + if frame.ResourceType == RntbdResourceTypeConnection { + responseFrame := buildContextFrame(frame) + _, err := writer.Write(responseFrame) + writer.Flush() + if err != nil { + logger.ErrorLn("Failed to write response frame:", err) + continue + } + continue + } else if frame.ResourceType == RntbdResourceTypeDatabase || + frame.ResourceType == RntbdResourceTypeCollection || + frame.ResourceType == RntbdResourceTypeDocument { + responseFrameBytes := s.passToApiServer(frame) + _, err := writer.Write(responseFrameBytes) + writer.Flush() + if err != nil { + logger.ErrorLn("Failed to write response frame:", err) + continue + } + continue + } else { + logger.Errorf("Received Unhandled RNTBD request from: %s with resource type: %s\n", conn.RemoteAddr(), frame.ResourceType.String()) + } + } +} + +func (s *RntbdServer) passToApiServer(frame *RntbdFrame) []byte { + req := frame.ToHttpRequest() + responseWriter := httptest.NewRecorder() + s.apiServer.GetRouter().ServeHTTP(responseWriter, req) + + responseFrameBuilder := ToRntbdResponseFrame(responseWriter) + responseFrameBuilder.SetActivityId(frame.ActivityId) + if transportRequestId, ok := frame.RequestHeaders[RntbdRequestHeaderTransportRequestID]; ok { + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderTransportRequestID), RntbdTokenTypeULong, transportRequestId) + } + + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderItemLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderLocalLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderGlobalCommittedLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderItemLocalLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderQuorumAckedLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderQuorumAckedLocalLSN), RntbdTokenTypeLongLong, int64(420)) + responseFrameBuilder.AddHeader(uint16(RntbdResponseHeaderCurrentReplicaSetSize), RntbdTokenTypeULong, uint32(1)) + + responseFrame := responseFrameBuilder.Build() + responseFrameBytes := responseFrame.ToBytes() + + return responseFrameBytes +}