|
| 1 | +// Copyright 2025 MongoDB Inc |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +package resource |
| 16 | + |
| 17 | +import "go.mongodb.org/atlas-sdk/v20250312010/admin" |
| 18 | + |
| 19 | +func NewStreamWorkspaceCreateReq(model *Model) *admin.StreamsTenant { |
| 20 | + if model == nil { |
| 21 | + return nil |
| 22 | + } |
| 23 | + dataProcessRegion := *model.DataProcessRegion |
| 24 | + streamTenant := &admin.StreamsTenant{ |
| 25 | + Name: model.WorkspaceName, |
| 26 | + GroupId: model.ProjectId, |
| 27 | + DataProcessRegion: &admin.StreamsDataProcessRegion{ |
| 28 | + CloudProvider: *dataProcessRegion.CloudProvider, |
| 29 | + Region: *dataProcessRegion.Region, |
| 30 | + }, |
| 31 | + } |
| 32 | + if streamConfig := model.StreamConfig; streamConfig != nil { |
| 33 | + streamTenant.StreamConfig = &admin.StreamConfig{} |
| 34 | + if tier := streamConfig.Tier; tier != nil { |
| 35 | + streamTenant.StreamConfig.Tier = tier |
| 36 | + } |
| 37 | + if maxTierSize := streamConfig.MaxTierSize; maxTierSize != nil { |
| 38 | + streamTenant.StreamConfig.MaxTierSize = maxTierSize |
| 39 | + } |
| 40 | + } |
| 41 | + return streamTenant |
| 42 | +} |
| 43 | + |
| 44 | +func newModelDataRegion(dataProcessRegion *admin.StreamsDataProcessRegion) *StreamsDataProcessRegion { |
| 45 | + return &StreamsDataProcessRegion{ |
| 46 | + CloudProvider: &dataProcessRegion.CloudProvider, |
| 47 | + Region: &dataProcessRegion.Region, |
| 48 | + } |
| 49 | +} |
| 50 | + |
| 51 | +func newModelStreamConfig(streamConfig *admin.StreamConfig) *StreamConfig { |
| 52 | + if streamConfig == nil { |
| 53 | + return nil |
| 54 | + } |
| 55 | + modelConfig := &StreamConfig{} |
| 56 | + if streamConfig.Tier != nil { |
| 57 | + modelConfig.Tier = streamConfig.Tier |
| 58 | + } |
| 59 | + if streamConfig.MaxTierSize != nil { |
| 60 | + modelConfig.MaxTierSize = streamConfig.MaxTierSize |
| 61 | + } |
| 62 | + return modelConfig |
| 63 | +} |
| 64 | + |
| 65 | +func newModelDBRoleToExecute(dbRole *admin.DBRoleToExecute) *DBRoleToExecute { |
| 66 | + return &DBRoleToExecute{ |
| 67 | + Role: dbRole.Role, |
| 68 | + Type: dbRole.Type, |
| 69 | + } |
| 70 | +} |
| 71 | + |
| 72 | +func newModelAuthentication(authentication *admin.StreamsKafkaAuthentication) *StreamsKafkaAuthentication { |
| 73 | + return &StreamsKafkaAuthentication{ |
| 74 | + Mechanism: authentication.Mechanism, |
| 75 | + Username: authentication.Username, |
| 76 | + } |
| 77 | +} |
| 78 | + |
| 79 | +func newModelSecurity(security *admin.StreamsKafkaSecurity) *StreamsKafkaSecurity { |
| 80 | + return &StreamsKafkaSecurity{ |
| 81 | + BrokerPublicCertificate: security.BrokerPublicCertificate, |
| 82 | + Protocol: security.Protocol, |
| 83 | + } |
| 84 | +} |
| 85 | + |
| 86 | +func NewModelConnections(streamConfig *[]admin.StreamsConnection) []StreamsConnection { |
| 87 | + if streamConfig == nil || len(*streamConfig) == 0 { |
| 88 | + return nil |
| 89 | + } |
| 90 | + |
| 91 | + connections := make([]StreamsConnection, 0) |
| 92 | + for _, connection := range *streamConfig { |
| 93 | + modelConnection := StreamsConnection{ |
| 94 | + Name: connection.Name, |
| 95 | + Type: connection.Type, |
| 96 | + } |
| 97 | + if connection.GetType() == Kafka { |
| 98 | + modelConnection.BootstrapServers = connection.BootstrapServers |
| 99 | + modelConnection.Authentication = newModelAuthentication(connection.Authentication) |
| 100 | + modelConnection.Security = newModelSecurity(connection.Security) |
| 101 | + } else if connection.GetType() == Cluster { |
| 102 | + modelConnection.ClusterName = connection.ClusterName |
| 103 | + modelConnection.DbRoleToExecute = newModelDBRoleToExecute(connection.DbRoleToExecute) |
| 104 | + } |
| 105 | + connections = append(connections, modelConnection) |
| 106 | + } |
| 107 | + return connections |
| 108 | +} |
| 109 | + |
| 110 | +func GetStreamWorkspaceModel(streamTenant *admin.StreamsTenant, currentModel *Model) *Model { |
| 111 | + model := new(Model) |
| 112 | + |
| 113 | + if currentModel != nil { |
| 114 | + model = currentModel |
| 115 | + } |
| 116 | + |
| 117 | + if streamTenant != nil { |
| 118 | + model.WorkspaceName = streamTenant.Name |
| 119 | + model.DataProcessRegion = newModelDataRegion(streamTenant.DataProcessRegion) |
| 120 | + model.StreamConfig = newModelStreamConfig(streamTenant.StreamConfig) |
| 121 | + model.ProjectId = streamTenant.GroupId |
| 122 | + model.Id = streamTenant.Id |
| 123 | + model.Hostnames = streamTenant.GetHostnames() |
| 124 | + model.Connections = NewModelConnections(streamTenant.Connections) |
| 125 | + } |
| 126 | + |
| 127 | + return model |
| 128 | +} |
0 commit comments