Merge pull request #4304 from ViacheslavKlimov/feature/snmp-refactoring

SNMP refactoring

application/src/main/resources/logback.xml

@@ -26,6 +26,7 @@
     </appender>
 
     <logger name="org.thingsboard.server" level="INFO" />
+    <logger name="org.thingsboard.server.transport.snmp" level="TRACE" />
 
 <!--    <logger name="org.thingsboard.server.service.queue" level="TRACE" />-->
 <!--    <logger name="org.thingsboard.server.service.transport" level="TRACE" />-->

application/src/main/resources/thingsboard.yml

@@ -648,6 +648,11 @@ transport:
     redis.enabled: "${LWM2M_REDIS_ENABLED:false}"
   snmp:
     enabled: "${SNMP_ENABLED:true}"
+    response_processing:
+      # parallelism level for executor (workStealingPool) that is responsible for handling responses from SNMP devices
+      parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:20}"
+    # to configure SNMP to work over UDP or TCP
+    underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}"
 
 swagger:
   api_path_regex: "${SWAGGER_API_PATH_REGEX:/api.*}"

common/data/src/main/java/org/thingsboard/server/common/data/device/data/SnmpDeviceTransportConfiguration.java

@@ -25,8 +25,10 @@ import org.thingsboard.server.common.data.transport.snmp.SnmpProtocolVersion;
 public class SnmpDeviceTransportConfiguration implements DeviceTransportConfiguration {
     private String address;
     private int port;
-    private String community;
     private SnmpProtocolVersion protocolVersion;
+    private String securityName;
+    private String authenticationPassphrase; // for SNMP v3
+    private String privacyPassphrase; // for SNMP v3
 
     @Override
     public DeviceTransportType getType() {
@@ -43,6 +45,6 @@ public class SnmpDeviceTransportConfiguration implements DeviceTransportConfigur
     @JsonIgnore
     private boolean isValid() {
         return StringUtils.isNotBlank(address) && port > 0 &&
-                StringUtils.isNotBlank(community) && protocolVersion != null;
+                StringUtils.isNotBlank(securityName) && protocolVersion != null;
     }
 }

common/data/src/main/java/org/thingsboard/server/common/data/device/profile/SnmpDeviceProfileTransportConfiguration.java

@@ -18,35 +18,21 @@ package org.thingsboard.server.common.data.device.profile;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import lombok.Data;
 import org.thingsboard.server.common.data.DeviceTransportType;
-import org.thingsboard.server.common.data.transport.snmp.SnmpMapping;
+import org.thingsboard.server.common.data.transport.snmp.configs.SnmpCommunicationConfig;
 
-import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 @Data
 public class SnmpDeviceProfileTransportConfiguration implements DeviceProfileTransportConfiguration {
-    private int pollPeriodMs;
     private int timeoutMs;
     private int retries;
-    private List<SnmpMapping> telemetryMappings;
-    private List<SnmpMapping> attributesMappings;
+    private List<SnmpCommunicationConfig> communicationConfigs;
 
     @Override
     public DeviceTransportType getType() {
         return DeviceTransportType.SNMP;
     }
 
-    @JsonIgnore
-    public List<SnmpMapping> getAllMappings() {
-        if (telemetryMappings != null && attributesMappings != null) {
-            return Stream.concat(telemetryMappings.stream(), attributesMappings.stream()).collect(Collectors.toList());
-        } else {
-            return Collections.emptyList();
-        }
-    }
-
     @Override
     public void validate() {
         if (!isValid()) {
@@ -56,9 +42,6 @@ public class SnmpDeviceProfileTransportConfiguration implements DeviceProfileTra
 
     @JsonIgnore
     private boolean isValid() {
-        List<SnmpMapping> mappings = getAllMappings();
-        return pollPeriodMs > 0 && timeoutMs > 0 && retries >= 0 &&
-                !mappings.isEmpty() && mappings.stream().allMatch(SnmpMapping::isValid) &&
-                mappings.stream().map(SnmpMapping::getOid).distinct().count() == mappings.size();
+        return true;
     }
 }

common/data/src/main/java/org/thingsboard/server/common/data/device/JsonBasedTransportConfiguration.java → common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpCommunicationSpec.java

@@ -13,27 +13,28 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.common.data.device;
+package org.thingsboard.server.common.data.transport.snmp;
 
-import com.fasterxml.jackson.annotation.JsonAnyGetter;
-import com.fasterxml.jackson.annotation.JsonAnySetter;
-import com.fasterxml.jackson.annotation.JsonIgnore;
+public enum SnmpCommunicationSpec {
+    TELEMETRY_QUERYING(true),
+    CLIENT_ATTRIBUTES_QUERYING(true),
 
-import java.util.HashMap;
-import java.util.Map;
+    SHARED_ATTRIBUTES_SETTING,
 
-public abstract class JsonBasedTransportConfiguration {
+    TELEMETRY_TRAPS_RECEIVING,
+    CLIENT_ATTRIBUTES_TRAPS_RECEIVING;
 
-    @JsonIgnore
-    private final Map<String, Object> properties = new HashMap<>();
+    private final boolean isRepeatingQuerying;
 
-    @JsonAnyGetter
-    public Map<String, Object> properties() {
-        return this.properties;
+    SnmpCommunicationSpec() {
+        this.isRepeatingQuerying = false;
     }
 
-    @JsonAnySetter
-    public void put(String name, Object value) {
-        this.properties.put(name, value);
+    SnmpCommunicationSpec(boolean isRepeatingQuerying) {
+        this.isRepeatingQuerying = isRepeatingQuerying;
+    }
+
+    public boolean isRepeatingQuerying() {
+        return isRepeatingQuerying;
     }
 }

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/SnmpMapping.java

@@ -25,7 +25,6 @@ import java.util.regex.Pattern;
 @Data
 public class SnmpMapping {
     private String oid;
-    private SnmpMethod method;
     private String key;
     private DataType dataType;
 
@@ -33,7 +32,7 @@ public class SnmpMapping {
 
     @JsonIgnore
     public boolean isValid() {
-        return StringUtils.isNotEmpty(oid) && OID_PATTERN.matcher(oid).matches() && method != null &&
+        return StringUtils.isNotEmpty(oid) && OID_PATTERN.matcher(oid).matches() &&
                 StringUtils.isNotBlank(key) && dataType != null;
     }
 }

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesQueryingSnmpCommunicationConfig.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+
+public class ClientAttributesQueryingSnmpCommunicationConfig extends RepeatingQueryingSnmpCommunicationConfig {
+    @Override
+    public SnmpCommunicationSpec getSpec() {
+        return SnmpCommunicationSpec.CLIENT_ATTRIBUTES_QUERYING;
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/ClientAttributesTrapsReceivingSnmpCommunicationConfig.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+
+public class ClientAttributesTrapsReceivingSnmpCommunicationConfig extends SnmpCommunicationConfig {
+    @Override
+    public SnmpCommunicationSpec getSpec() {
+        return SnmpCommunicationSpec.CLIENT_ATTRIBUTES_TRAPS_RECEIVING;
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/RepeatingQueryingSnmpCommunicationConfig.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.thingsboard.server.common.data.transport.snmp.SnmpMethod;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+public abstract class RepeatingQueryingSnmpCommunicationConfig extends SnmpCommunicationConfig {
+    private Long queryingFrequencyMs;
+
+    @Override
+    public SnmpMethod getMethod() {
+        return SnmpMethod.GET;
+    }
+
+    @Override
+    public boolean isValid() {
+        return true;
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SharedAttributesSettingSnmpCommunicationConfig.java

@@ -0,0 +1,36 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+import org.thingsboard.server.common.data.transport.snmp.SnmpMethod;
+
+public class SharedAttributesSettingSnmpCommunicationConfig extends SnmpCommunicationConfig {
+    @Override
+    public SnmpCommunicationSpec getSpec() {
+        return SnmpCommunicationSpec.SHARED_ATTRIBUTES_SETTING;
+    }
+
+    @Override
+    public SnmpMethod getMethod() {
+        return SnmpMethod.SET;
+    }
+
+    @Override
+    public boolean isValid() {
+        return true;
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/SnmpCommunicationConfig.java

@@ -0,0 +1,62 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+import org.thingsboard.server.common.data.transport.snmp.SnmpMapping;
+import org.thingsboard.server.common.data.transport.snmp.SnmpMethod;
+
+import java.util.List;
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "spec")
+@JsonSubTypes({
+        @Type(value = TelemetryQueryingSnmpCommunicationConfig.class, name = "TELEMETRY_QUERYING"),
+        @Type(value = ClientAttributesQueryingSnmpCommunicationConfig.class, name = "CLIENT_ATTRIBUTES_QUERYING"),
+        @Type(value = SharedAttributesSettingSnmpCommunicationConfig.class, name = "SHARED_ATTRIBUTES_SETTING"),
+        @Type(value = TelemetryTrapsReceivingSnmpCommunicationConfig.class, name = "TELEMETRY_TRAPS_RECEIVING"),
+        @Type(value = ClientAttributesTrapsReceivingSnmpCommunicationConfig.class, name = "CLIENT_ATTRIBUTES_TRAPS_RECEIVING")
+})
+public abstract class SnmpCommunicationConfig {
+    protected List<SnmpMapping> mappings;
+
+    public List<SnmpMapping> getMappings() {
+        return mappings;
+    }
+
+    public abstract SnmpCommunicationSpec getSpec();
+
+    @JsonIgnore
+    public SnmpMethod getMethod() {
+        return null;
+    }
+
+    @JsonIgnore
+    public boolean isValid() {
+        return true;
+    }
+
+    public void validate() {
+        if (!isValid()) {
+            throw new IllegalArgumentException("Communication config is not valid");
+        }
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryQueryingSnmpCommunicationConfig.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class TelemetryQueryingSnmpCommunicationConfig extends RepeatingQueryingSnmpCommunicationConfig {
+    @Override
+    public SnmpCommunicationSpec getSpec() {
+        return SnmpCommunicationSpec.TELEMETRY_QUERYING;
+    }
+}

common/data/src/main/java/org/thingsboard/server/common/data/transport/snmp/configs/TelemetryTrapsReceivingSnmpCommunicationConfig.java

@@ -0,0 +1,30 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.transport.snmp.configs;
+
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+
+public class TelemetryTrapsReceivingSnmpCommunicationConfig extends SnmpCommunicationConfig {
+    @Override
+    public SnmpCommunicationSpec getSpec() {
+        return SnmpCommunicationSpec.TELEMETRY_TRAPS_RECEIVING;
+    }
+
+    @Override
+    public boolean isValid() {
+        return false;
+    }
+}

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mTransportServiceImpl.java

@@ -717,7 +717,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
      * @return - value of Resource or null
      */
     private String getResourceValueToString(LwM2mClient lwM2MClient, String path) {
-        LwM2mPath pathIds =new LwM2mPath(convertToObjectIdFromIdVer(path));
+        LwM2mPath pathIds = new LwM2mPath(convertToObjectIdFromIdVer(path));
         ResourceValue resourceValue = this.returnResourceValueFromLwM2MClient(lwM2MClient, path);
         return resourceValue == null ? null :
                 this.converter.convertValue(resourceValue.getResourceValue(), this.lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModelType(lwM2MClient.getRegistration(), pathIds), ResourceModel.Type.STRING, pathIds).toString();
@@ -831,8 +831,10 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
 
             // #5.1
             if (!observeOld.equals(observeNew)) {
-                Set<String> observeSetOld = new Gson().fromJson(observeOld, new TypeToken<Set<String>>() {}.getType());
-                Set<String> observeSetNew = new Gson().fromJson(observeNew, new TypeToken<Set<String>>() {}.getType());
+                Set<String> observeSetOld = new Gson().fromJson(observeOld, new TypeToken<Set<String>>() {
+                }.getType());
+                Set<String> observeSetNew = new Gson().fromJson(observeNew, new TypeToken<Set<String>>() {
+                }.getType());
                 //#5.2 add
                 //  path Attr/Telemetry includes newObserve
                 attributeSetOld.addAll(telemetrySetOld);
@@ -968,7 +970,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
         LwM2mClientProfile profile = lwM2mClientContext.getProfile(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
         Registration registration = lwM2mClientContext.getLwM2MClient(sessionInfo).getRegistration();
         return profile.getPostKeyNameProfile().getAsJsonObject().entrySet().stream()
-                .filter(e -> e.getValue().getAsString().equals(name) && validateResourceInModelByVer(registration, e.getKey())).findFirst().map(Map.Entry::getKey)
+                .filter(e -> e.getValue().getAsString().equals(name) && validateResourceInModel(registration, e.getKey(), false)).findFirst().map(Map.Entry::getKey)
                 .orElse("");
     }
 
@@ -988,7 +990,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
         try {
             LwM2mClient lwM2MClient = lwM2mClientContext.getLwM2MClient(sessionInfo);
             attributesResponse.getSharedAttributeListList().forEach(attr -> {
-                String path =this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey());
+                String path = this.getPathAttributeUpdate(sessionInfo, attr.getKv().getKey());
                 if (path != null) {
                     // #1.1
                     if (lwM2MClient.getDelayedRequests().containsKey(path) && attr.getTs() > lwM2MClient.getDelayedRequests().get(path).getTs()) {
@@ -1117,7 +1119,7 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
 
         ConcurrentMap<String, String> keyNamesIsWritable = keyNamesMap.entrySet()
                 .stream()
-                .filter(e -> (attrSet.contains(e.getKey()) && resourceIsWritable(lwM2MClient.getRegistration(), e.getKey())))
+                .filter(e -> (attrSet.contains(e.getKey()) && validateResourceInModel(lwM2MClient.getRegistration(), e.getKey(), true)))
                 .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue));
 
         Set<String> namesIsWritable = ConcurrentHashMap.newKeySet();
@@ -1130,17 +1132,13 @@ public class LwM2mTransportServiceImpl implements LwM2mTransportService {
         return "LWM2M";
     }
 
-    private boolean resourceIsWritable(Registration registration, String pathKey) {
-        return validateResourceInModelByVer(registration, pathKey) &&
-                lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration,
-                        new LwM2mPath(convertToObjectIdFromIdVer(pathKey))).operations.isWritable();
-    }
-
-    private boolean validateResourceInModelByVer(Registration registration, String pathKey) {
+    private boolean validateResourceInModel(Registration registration, String pathKey, boolean isWritable) {
+        ResourceModel resourceModel = lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration,
+                new LwM2mPath(convertToObjectIdFromIdVer(pathKey)));
         Integer objectId = validateObjectIdFromKey(pathKey);
         String objectVer = validateObjectVerFromKey(pathKey);
-        LwM2mPath lwM2mPath = new LwM2mPath(convertToObjectIdFromIdVer(pathKey));
-        return objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId)) &&
-                lwM2mTransportContextServer.getLwM2MTransportConfigServer().getResourceModel(registration, lwM2mPath) != null;
+        return resourceModel != null && (isWritable ?
+                objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId)) && resourceModel.operations.isWritable() :
+                objectId != null && objectVer != null && objectVer.equals(registration.getSupportedVersion(objectId)));
     }
 }

common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mVersionedModelProvider.java

@@ -22,6 +22,7 @@ import org.eclipse.leshan.core.model.ObjectModel;
 import org.eclipse.leshan.core.model.ResourceModel;
 import org.eclipse.leshan.server.model.LwM2mModelProvider;
 import org.eclipse.leshan.server.registration.Registration;
+import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClientContext;
 
@@ -30,6 +31,7 @@ import java.util.Base64;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Optional;
 
 import static org.thingsboard.server.common.data.ResourceType.LWM2M_MODEL;
 import static org.thingsboard.server.common.data.lwm2m.LwM2mConstants.LWM2M_SEPARATOR_KEY;
@@ -83,6 +85,7 @@ public class LwM2mVersionedModelProvider implements LwM2mModelProvider {
                 if (objectModel != null)
                     return objectModel.resources.get(resourceId);
                 else
+                    log.warn("TbResources (Object model) with id [{}/{}] not found on the server", objectId, resourceId);
                     return null;
             } catch (Exception e) {
                 log.error("", e);
@@ -116,17 +119,15 @@ public class LwM2mVersionedModelProvider implements LwM2mModelProvider {
 
         private ObjectModel getObjectModelDynamic(Integer objectId, String version) {
             String key = getKeyIdVer(objectId, version);
-            String xmlB64 = lwM2mTransportContextServer.getTransportResourceCache().get(
-                    this.tenantId,
-                    LWM2M_MODEL,
-                    key).
-                    getData();
-            return xmlB64 != null && !xmlB64.isEmpty() ?
-                    lwM2mTransportContextServer.parseFromXmlToObjectModel(
-                            Base64.getDecoder().decode(xmlB64),
-                            key + ".xml",
-                            new DefaultDDFFileValidator()) :
-                    null;
+
+            Optional<TbResource> tbResource = lwM2mTransportContextServer
+                    .getTransportResourceCache()
+                    .get(this.tenantId, LWM2M_MODEL, key);
+
+            return tbResource.map(resource -> lwM2mTransportContextServer.parseFromXmlToObjectModel(
+                    Base64.getDecoder().decode(resource.getData()),
+                    key + ".xml",
+                    new DefaultDDFFileValidator())).orElse(null);
         }
     }
 }

common/transport/snmp/pom.xml

@@ -58,5 +58,11 @@
             <groupId>org.snmp4j</groupId>
             <artifactId>snmp4j</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.snmp4j</groupId>
+            <artifactId>snmp4j-agent</artifactId>
+            <version>3.3.6</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulator.java

@@ -1,81 +0,0 @@
-/**
- * Copyright © 2016-2021 The Thingsboard Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.thingsboard.server.transport.snmp;
-
-import org.snmp4j.CommunityTarget;
-import org.snmp4j.PDU;
-import org.snmp4j.Snmp;
-import org.snmp4j.Target;
-import org.snmp4j.event.ResponseEvent;
-import org.snmp4j.mp.SnmpConstants;
-import org.snmp4j.smi.GenericAddress;
-import org.snmp4j.smi.OID;
-import org.snmp4j.smi.OctetString;
-import org.snmp4j.smi.VariableBinding;
-import org.snmp4j.transport.DefaultUdpTransportMapping;
-import org.snmp4j.transport.UdpTransportMapping;
-
-import java.io.IOException;
-
-/**
- * For testing purposes. Will be removed when the time comes
- */
-public class SnmpDeviceSimulator {
-    private final Target target;
-    private final OID oid = new OID(".1.3.6.1.2.1.1.1.0");
-    private Snmp snmp;
-
-    public SnmpDeviceSimulator(int port) {
-        String address = "udp:127.0.0.1/" + port;
-
-        CommunityTarget target = new CommunityTarget();
-        target.setCommunity(new OctetString("public"));
-        target.setAddress(GenericAddress.parse(address));
-        target.setRetries(2);
-        target.setTimeout(1500);
-        target.setVersion(SnmpConstants.version2c);
-
-        this.target = target;
-    }
-
-    public static void main(String[] args) throws IOException {
-        SnmpDeviceSimulator deviceSimulator = new SnmpDeviceSimulator(161);
-
-        deviceSimulator.start();
-        String response = deviceSimulator.sendRequest(PDU.GET);
-
-        System.out.println(response);
-    }
-
-    public void start() throws IOException {
-        UdpTransportMapping transport = new DefaultUdpTransportMapping();
-        transport.addTransportListener((sourceTransport, incomingAddress, wholeMessage, tmStateReference) -> {
-            System.out.println();
-        });
-        snmp = new Snmp(transport);
-
-        transport.listen();
-    }
-
-    public String sendRequest(int pduType) throws IOException {
-        PDU pdu = new PDU();
-        pdu.add(new VariableBinding(oid));
-        pdu.setType(pduType);
-
-        ResponseEvent responseEvent = snmp.send(pdu, target);
-        return responseEvent.getResponse().get(0).getVariable().toString();
-    }
-}

common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/SnmpTransportContext.java

@@ -17,9 +17,7 @@ package org.thingsboard.server.transport.snmp;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.snmp4j.PDU;
-import org.snmp4j.smi.OID;
-import org.snmp4j.smi.VariableBinding;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
 import org.thingsboard.server.common.data.Device;
@@ -32,8 +30,6 @@ import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.security.DeviceCredentials;
 import org.thingsboard.server.common.data.security.DeviceCredentialsType;
-import org.thingsboard.server.common.data.transport.snmp.SnmpMapping;
-import org.thingsboard.server.common.data.transport.snmp.SnmpMethod;
 import org.thingsboard.server.common.transport.DeviceUpdatedEvent;
 import org.thingsboard.server.common.transport.TransportContext;
 import org.thingsboard.server.common.transport.TransportDeviceProfileCache;
@@ -41,7 +37,6 @@ import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.common.transport.auth.SessionInfoCreator;
 import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
-import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto;
 import org.thingsboard.server.queue.util.AfterStartUp;
@@ -51,9 +46,7 @@ import org.thingsboard.server.transport.snmp.service.SnmpTransportBalancingServi
 import org.thingsboard.server.transport.snmp.service.SnmpTransportService;
 import org.thingsboard.server.transport.snmp.session.DeviceSessionContext;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -75,10 +68,11 @@ public class SnmpTransportContext extends TransportContext {
     private final SnmpTransportBalancingService balancingService;
 
     private final Map<DeviceId, DeviceSessionContext> sessions = new ConcurrentHashMap<>();
-    private final Map<DeviceProfileId, SnmpDeviceProfileTransportConfiguration> profilesTransportConfigs = new ConcurrentHashMap<>();
-    private final Map<DeviceProfileId, List<PDU>> profilesPdus = new ConcurrentHashMap<>();
     private Collection<DeviceId> allSnmpDevicesIds = new ConcurrentLinkedDeque<>();
 
+    @Value("${transport.snmp.underlying_protocol}")
+    private String snmpUnderlyingProtocol;
+
     @AfterStartUp(order = 2)
     public void initDevicesSessions() {
         log.info("Initializing SNMP devices sessions");
@@ -120,22 +114,19 @@ public class SnmpTransportContext extends TransportContext {
         SnmpDeviceProfileTransportConfiguration profileTransportConfiguration = (SnmpDeviceProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration();
         SnmpDeviceTransportConfiguration deviceTransportConfiguration = (SnmpDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration();
 
-        profilesTransportConfigs.put(deviceProfileId, profileTransportConfiguration);
-        profilesPdus.computeIfAbsent(deviceProfileId, id -> createPdus(profileTransportConfiguration));
-
         DeviceSessionContext deviceSessionContext = new DeviceSessionContext(
-                device, deviceProfile,
-                credentials.getCredentialsId(), deviceTransportConfiguration,
-                this, snmpTransportService
+                device, deviceProfile, credentials.getCredentialsId(),
+                profileTransportConfiguration, deviceTransportConfiguration,
+                this, snmpTransportService, snmpUnderlyingProtocol
         );
         registerSessionMsgListener(deviceSessionContext);
         sessions.put(device.getId(), deviceSessionContext);
+        snmpTransportService.createQueryingTasks(deviceSessionContext);
         log.info("Established SNMP device session for device {}", device.getId());
     }
 
     private void updateDeviceSession(DeviceSessionContext sessionContext, Device device, DeviceProfile deviceProfile) {
         log.info("Updating SNMP device session for device {}", device.getId());
-        DeviceProfileId deviceProfileId = deviceProfile.getId();
 
         DeviceCredentials credentials = protoEntityService.getDeviceCredentialsByDeviceId(device.getId());
         if (credentials.getCredentialsType() != DeviceCredentialsType.ACCESS_TOKEN) {
@@ -148,14 +139,13 @@ public class SnmpTransportContext extends TransportContext {
         SnmpDeviceTransportConfiguration newDeviceTransportConfiguration = (SnmpDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration();
 
         if (!newProfileTransportConfiguration.equals(sessionContext.getProfileTransportConfiguration())) {
-            profilesPdus.put(deviceProfileId, createPdus(newProfileTransportConfiguration));
-            profilesTransportConfigs.put(deviceProfileId, newProfileTransportConfiguration);
-
             sessionContext.setProfileTransportConfiguration(newProfileTransportConfiguration);
-            sessionContext.initTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration);
+            sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration);
+            snmpTransportService.cancelQueryingTasks(sessionContext);
+            snmpTransportService.createQueryingTasks(sessionContext);
         } else if (!newDeviceTransportConfiguration.equals(sessionContext.getDeviceTransportConfiguration())) {
             sessionContext.setDeviceTransportConfiguration(newDeviceTransportConfiguration);
-            sessionContext.initTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration);
+            sessionContext.initializeTarget(newProfileTransportConfiguration, newDeviceTransportConfiguration);
         } else {
             log.trace("Configuration of the device {} was not updated", device);
         }
@@ -167,22 +157,14 @@ public class SnmpTransportContext extends TransportContext {
         sessionContext.close();
         transportService.deregisterSession(sessionContext.getSessionInfo());
         sessions.remove(sessionContext.getDeviceId());
-        log.trace("Deregistered and removed session");
-
-        DeviceProfileId deviceProfileId = sessionContext.getDeviceProfile().getId();
-        if (sessions.values().stream()
-                .map(DeviceAwareSessionContext::getDeviceProfile)
-                .noneMatch(deviceProfile -> deviceProfile.getId().equals(deviceProfileId))) {
-            log.trace("Removed values for device profile {} from configs and pdus caches", deviceProfileId);
-            profilesTransportConfigs.remove(deviceProfileId);
-            profilesPdus.remove(deviceProfileId);
-        }
+        snmpTransportService.cancelQueryingTasks(sessionContext);
+        log.trace("Unregistered and removed session");
     }
 
     private void registerSessionMsgListener(DeviceSessionContext deviceSessionContext) {
         transportService.process(DeviceTransportType.SNMP,
                 TransportProtos.ValidateDeviceTokenRequestMsg.newBuilder().setToken(deviceSessionContext.getToken()).build(),
-                new TransportServiceCallback<ValidateDeviceCredentialsResponse>() {
+                new TransportServiceCallback<>() {
                     @Override
                     public void onSuccess(ValidateDeviceCredentialsResponse msg) {
                         if (msg.hasDeviceInfo()) {
@@ -191,6 +173,7 @@ public class SnmpTransportContext extends TransportContext {
                             );
 
                             transportService.registerAsyncSession(sessionInfo, deviceSessionContext);
+                            transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null);
                             deviceSessionContext.setSessionInfo(sessionInfo);
                             deviceSessionContext.setDeviceInfo(msg.getDeviceInfo());
                         } else {
@@ -205,23 +188,6 @@ public class SnmpTransportContext extends TransportContext {
                 });
     }
 
-    private List<PDU> createPdus(SnmpDeviceProfileTransportConfiguration deviceProfileConfig) {
-        Map<SnmpMethod, List<VariableBinding>> bindingsPerMethod = new HashMap<>();
-
-        deviceProfileConfig.getAllMappings().forEach(mapping -> bindingsPerMethod
-                .computeIfAbsent(mapping.getMethod(), v -> new ArrayList<>())
-                .add(new VariableBinding(new OID(mapping.getOid()))));
-
-        return bindingsPerMethod.keySet().stream()
-                .map(method -> {
-                    PDU request = new PDU();
-                    request.setType(method.getCode());
-                    request.addAll(bindingsPerMethod.get(method));
-                    return request;
-                })
-                .collect(Collectors.toList());
-    }
-
     @EventListener(DeviceUpdatedEvent.class)
     public void onDeviceUpdatedOrCreated(DeviceUpdatedEvent deviceUpdatedEvent) {
         Device device = deviceUpdatedEvent.getDevice();
@@ -293,27 +259,4 @@ public class SnmpTransportContext extends TransportContext {
         return sessions.values();
     }
 
-    public Map<DeviceProfileId, List<PDU>> getProfilesPdus() {
-        return profilesPdus;
-    }
-
-    public Optional<SnmpMapping> getAttributeMapping(DeviceProfileId deviceProfileId, OID responseOid) {
-        if (profilesTransportConfigs.containsKey(deviceProfileId)) {
-            return getMapping(responseOid, profilesTransportConfigs.get(deviceProfileId).getAttributesMappings());
-        }
-        return Optional.empty();
-    }
-
-    public Optional<SnmpMapping> getTelemetryMapping(DeviceProfileId deviceProfileId, OID responseOid) {
-        if (profilesTransportConfigs.containsKey(deviceProfileId)) {
-            return getMapping(responseOid, profilesTransportConfigs.get(deviceProfileId).getTelemetryMappings());
-        }
-        return Optional.empty();
-    }
-
-    private Optional<SnmpMapping> getMapping(OID oid, List<SnmpMapping> mappings) {
-        return mappings.stream()
-                .filter(mapping -> new OID(mapping.getOid()).equals(oid))
-                .findFirst();
-    }
 }

common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/service/SnmpTransportService.java

@@ -16,24 +16,38 @@
 package org.thingsboard.server.transport.snmp.service;
 
 import com.google.gson.JsonObject;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.snmp4j.CommandResponder;
+import org.snmp4j.CommandResponderEvent;
 import org.snmp4j.PDU;
 import org.snmp4j.Snmp;
+import org.snmp4j.TransportMapping;
 import org.snmp4j.event.ResponseEvent;
 import org.snmp4j.smi.Null;
+import org.snmp4j.smi.OID;
+import org.snmp4j.smi.OctetString;
+import org.snmp4j.smi.TcpAddress;
+import org.snmp4j.smi.UdpAddress;
 import org.snmp4j.smi.VariableBinding;
+import org.snmp4j.transport.DefaultTcpTransportMapping;
 import org.snmp4j.transport.DefaultUdpTransportMapping;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.common.data.TbTransportService;
 import org.thingsboard.server.common.data.id.DeviceProfileId;
 import org.thingsboard.server.common.data.kv.DataType;
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+import org.thingsboard.server.common.data.transport.snmp.SnmpMapping;
+import org.thingsboard.server.common.data.transport.snmp.configs.RepeatingQueryingSnmpCommunicationConfig;
+import org.thingsboard.server.common.data.transport.snmp.configs.SnmpCommunicationConfig;
 import org.thingsboard.server.common.transport.TransportService;
 import org.thingsboard.server.common.transport.TransportServiceCallback;
 import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.gen.transport.TransportProtos;
-import org.thingsboard.server.queue.util.AfterStartUp;
 import org.thingsboard.server.queue.util.TbSnmpTransportComponent;
 import org.thingsboard.server.transport.snmp.SnmpTransportContext;
 import org.thingsboard.server.transport.snmp.session.DeviceSessionContext;
@@ -41,21 +55,38 @@ import org.thingsboard.server.transport.snmp.session.DeviceSessionContext;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
 import java.io.IOException;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 @TbSnmpTransportComponent
 @Service
 @Slf4j
-public class SnmpTransportService implements TbTransportService {
+public class SnmpTransportService implements TbTransportService, CommandResponder {
     private final SnmpTransportContext snmpTransportContext;
     private final TransportService transportService;
 
+    @Getter
     private Snmp snmp;
-    private ScheduledExecutorService pollingExecutor;
-    private ExecutorService snmpResponseProcessingExecutor;
+    private ScheduledExecutorService queryingExecutor;
+    private ExecutorService responseProcessingExecutor;
+
+    private final Map<SnmpCommunicationSpec, BiConsumer<JsonObject, DeviceSessionContext>> responseProcessors = new EnumMap<>(SnmpCommunicationSpec.class);
+
+    @Value("${transport.snmp.response_processing.parallelism_level}")
+    private Integer responseProcessingParallelismLevel;
+    @Value("${transport.snmp.underlying_protocol}")
+    private String snmpUnderlyingProtocol;
 
     public SnmpTransportService(@Lazy SnmpTransportContext snmpTransportContext,
                                 TransportService transportService) {
@@ -67,56 +98,144 @@ public class SnmpTransportService implements TbTransportService {
     private void init() throws IOException {
         log.info("Initializing SNMP transport service");
 
-        pollingExecutor = Executors.newScheduledThreadPool(1, ThingsBoardThreadFactory.forName("snmp-polling"));
-        //TODO: Set parallelism value in the config
-        snmpResponseProcessingExecutor = Executors.newWorkStealingPool(20);
+        queryingExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("snmp-querying"));
+        responseProcessingExecutor = Executors.newWorkStealingPool(responseProcessingParallelismLevel);
 
         initializeSnmp();
+        initializeTrapsListener();
+        configureResponseProcessors();
 
         log.info("SNMP transport service initialized");
     }
 
     private void initializeSnmp() throws IOException {
-        snmp = new Snmp(new DefaultUdpTransportMapping());
+        TransportMapping<?> transportMapping;
+        switch (snmpUnderlyingProtocol) {
+            case "udp":
+                transportMapping = new DefaultUdpTransportMapping();
+                break;
+            case "tcp":
+                transportMapping = new DefaultTcpTransportMapping();
+                break;
+            default:
+                throw new IllegalArgumentException("Underlying protocol " + snmpUnderlyingProtocol + " for SNMP is not supported");
+        }
+        snmp = new Snmp(transportMapping);
         snmp.listen();
     }
 
-    @AfterStartUp(order = 10)
-    public void startPolling() {
-        log.info("Starting SNMP polling");
-        //TODO: Get poll period from configuration;
-        int pollPeriodSeconds = 1;
+    private void initializeTrapsListener() throws IOException {
+        int trapsListeningPort = 1062;
+        String bindingAddress = "0.0.0.0/" + trapsListeningPort;
 
-        pollingExecutor.scheduleWithFixedDelay(() -> {
-            snmpTransportContext.getSessions().forEach(this::executeSnmpRequest);
-        }, 0, pollPeriodSeconds, TimeUnit.SECONDS);
-    }
+        TransportMapping<?> transportMapping;
+        switch (snmpUnderlyingProtocol) {
+            case "udp":
+                transportMapping = new DefaultUdpTransportMapping(new UdpAddress(bindingAddress));
+                break;
+            case "tcp":
+                transportMapping = new DefaultTcpTransportMapping(new TcpAddress(bindingAddress));
+                break;
+            default:
+                throw new IllegalArgumentException("Underlying protocol " + snmpUnderlyingProtocol + " for SNMP is not supported");
+        }
+
+
+        Snmp trapsSnmp = new Snmp(transportMapping);
+        trapsSnmp.addCommandResponder(this);
 
-    private void executeSnmpRequest(DeviceSessionContext sessionContext) {
-        long timeNow = System.currentTimeMillis();
-        long nextRequestExecutionTime = sessionContext.getPreviousRequestExecutedAt() + sessionContext.getProfileTransportConfiguration().getPollPeriodMs();
+        transportMapping.listen();
+    }
 
-        if (nextRequestExecutionTime < timeNow) {
-            sessionContext.setPreviousRequestExecutedAt(timeNow);
+    public void createQueryingTasks(DeviceSessionContext sessionContext) {
+        List<ScheduledFuture<?>> queryingTasks = sessionContext.getProfileTransportConfiguration().getCommunicationConfigs().stream()
+                .filter(config -> config.getSpec().isRepeatingQuerying())
+                .map(config -> {
+                    RepeatingQueryingSnmpCommunicationConfig repeatingCommunicationConfig = (RepeatingQueryingSnmpCommunicationConfig) config;
+                    return createQueryingTaskForConfig(sessionContext, repeatingCommunicationConfig);
+                })
+                .collect(Collectors.toList());
+        sessionContext.setQueryingTasks(queryingTasks);
+    }
 
-            DeviceProfileId deviceProfileId = sessionContext.getDeviceProfile().getId();
-            snmpTransportContext.getProfilesPdus().get(deviceProfileId).forEach(pdu -> {
-                try {
-                    log.debug("[{}] Sending SNMP message for device {}", pdu.getRequestID(), sessionContext.getDeviceId());
-                    snmp.send(pdu, sessionContext.getTarget(), deviceProfileId, sessionContext);
-                } catch (Exception e) {
-                    log.error("Failed to send SNMP request: {}", e.getMessage());
+    private ScheduledFuture<?> createQueryingTaskForConfig(DeviceSessionContext sessionContext, RepeatingQueryingSnmpCommunicationConfig communicationConfig) {
+        Long queryingFrequency = communicationConfig.getQueryingFrequencyMs();
+        return queryingExecutor.scheduleWithFixedDelay(() -> {
+            try {
+                if (sessionContext.isActive()) {
+                    sendRequest(sessionContext, communicationConfig);
                 }
-            });
+            } catch (Exception e) {
+                log.error("Failed to send SNMP request for device {}: {}", sessionContext.getDeviceId(), e.getMessage());
+            }
+        }, queryingFrequency, queryingFrequency, TimeUnit.MILLISECONDS);
+    }
+
+    public void cancelQueryingTasks(DeviceSessionContext sessionContext) {
+        sessionContext.getQueryingTasks().forEach(task -> task.cancel(true));
+        sessionContext.getQueryingTasks().clear();
+    }
+
+    public void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig) throws IOException {
+        PDU request = createPdu(communicationConfig);
+        executeRequest(sessionContext, request);
+    }
+
+    public void sendRequest(DeviceSessionContext sessionContext, SnmpCommunicationConfig communicationConfig, Map<String, String> values) throws IOException {
+        PDU request = createPduWithValues(communicationConfig, values);
+        executeRequest(sessionContext, request);
+    }
+
+    private void executeRequest(DeviceSessionContext sessionContext, PDU request) throws IOException {
+        if (request.size() > 0) {
+            log.trace("Executing SNMP request for device {}. Variables bindings: {}", sessionContext.getDeviceId(), request.getVariableBindings());
+            snmp.send(request, sessionContext.getTarget(), sessionContext.getDeviceProfile().getId(), sessionContext);
         }
     }
 
-    public void onNewDeviceResponse(ResponseEvent responseEvent, DeviceSessionContext sessionContext) {
-        ((Snmp) responseEvent.getSource()).cancel(responseEvent.getRequest(), sessionContext);
-        snmpResponseProcessingExecutor.submit(() -> processSnmpResponse(responseEvent, sessionContext));
+    private PDU createPdu(SnmpCommunicationConfig communicationConfig) {
+        PDU pdu = new PDU();
+        pdu.setType(communicationConfig.getMethod().getCode());
+        pdu.addAll(communicationConfig.getMappings().stream()
+                .map(mapping -> new VariableBinding(new OID(mapping.getOid())))
+                .collect(Collectors.toList()));
+        return pdu;
     }
 
-    private void processSnmpResponse(ResponseEvent event, DeviceSessionContext sessionContext) {
+    private PDU createPduWithValues(SnmpCommunicationConfig communicationConfig, Map<String, String> values) {
+        PDU pdu = new PDU();
+        pdu.setType(communicationConfig.getMethod().getCode());
+        pdu.addAll(communicationConfig.getMappings().stream()
+                .filter(mapping -> values.containsKey(mapping.getKey()))
+                .map(mapping -> {
+                    String value = values.get(mapping.getKey());
+                    return new VariableBinding(new OID(mapping.getOid()), new OctetString(value));
+                })
+                .collect(Collectors.toList()));
+        return pdu;
+    }
+
+
+    private void processTrap(CommandResponderEvent event) {
+        if (event.getPDU().getType() != PDU.TRAP) return;
+
+        snmpTransportContext.getSessions().stream()
+                .filter(sessionContext -> {
+                    // TODO: SNMP v3 support
+                    return sessionContext.getTarget().getSecurityName().equals(OctetString.fromByteArray(event.getSecurityName())) &&
+                            sessionContext.getTarget().getAddress().equals(event.getPeerAddress());
+                })
+                .findFirst()
+                .ifPresentOrElse(sessionContext -> {
+                    responseProcessingExecutor.execute(() -> processResponse(sessionContext, event.getPDU()));
+                }, () -> {
+                    log.debug("SNMP event is from unknown source: {}", event);
+                });
+    }
+
+    public void processResponseEvent(DeviceSessionContext sessionContext, ResponseEvent event) {
+        ((Snmp) event.getSource()).cancel(event.getRequest(), sessionContext);
+
         if (event.getError() != null) {
             log.warn("Response error: {}", event.getError().getMessage(), event.getError());
             return;
@@ -127,16 +246,29 @@ public class SnmpTransportService implements TbTransportService {
             log.warn("No response from SNMP device {}, requestId: {}", sessionContext.getDeviceId(), event.getRequest().getRequestID());
             return;
         }
-
         DeviceProfileId deviceProfileId = (DeviceProfileId) event.getUserObject();
         log.debug("[{}] Processing SNMP response for device {} with device profile {}: {}",
                 response.getRequestID(), sessionContext.getDeviceId(), deviceProfileId, response);
 
-        JsonObject telemetry = new JsonObject();
-        JsonObject attributes = new JsonObject();
+        responseProcessingExecutor.execute(() -> processResponse(sessionContext, response));
+    }
 
-        for (int i = 0; i < response.size(); i++) {
-            VariableBinding variableBinding = response.get(i);
+    private void processResponse(DeviceSessionContext sessionContext, PDU responsePdu) {
+        Map<OID, SnmpMapping> mappings = new HashMap<>();
+        Map<OID, SnmpCommunicationConfig> configs = new HashMap<>();
+        Map<SnmpCommunicationSpec, JsonObject> responses = new EnumMap<>(SnmpCommunicationSpec.class);
+
+        for (SnmpCommunicationConfig config : sessionContext.getProfileTransportConfiguration().getCommunicationConfigs()) {
+            for (SnmpMapping mapping : config.getMappings()) {
+                OID oid = new OID(mapping.getOid());
+                mappings.put(oid, mapping);
+                configs.put(oid, config);
+            }
+            responses.put(config.getSpec(), new JsonObject());
+        }
+
+        for (int i = 0; i < responsePdu.size(); i++) {
+            VariableBinding variableBinding = responsePdu.get(i);
             log.trace("Processing variable binding {}: {}", i, variableBinding);
 
             if (variableBinding.getVariable() instanceof Null) {
@@ -144,37 +276,57 @@ public class SnmpTransportService implements TbTransportService {
                 continue;
             }
 
-            snmpTransportContext.getTelemetryMapping(deviceProfileId, variableBinding.getOid()).ifPresent(mapping -> {
-                log.trace("Found telemetry mapping for oid {}: {}", variableBinding.getOid(), mapping);
-                processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), telemetry);
-            });
+            OID oid = variableBinding.getOid();
+            if (!mappings.containsKey(oid)) {
+                log.debug("No SNMP mapping for oid {}", oid);
+                continue;
+            }
 
-            snmpTransportContext.getAttributeMapping(deviceProfileId, variableBinding.getOid()).ifPresent(mapping -> {
-                log.trace("Found attribute mapping for oid {}: {}", variableBinding.getOid(), mapping);
-                processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), attributes);
-            });
-        }
+            SnmpCommunicationSpec spec = configs.get(oid).getSpec();
+            if (!responseProcessors.containsKey(spec)) {
+                log.debug("No response processor found for spec {}", spec);
+                continue;
+            }
 
-        if (telemetry.entrySet().isEmpty() && attributes.entrySet().isEmpty()) {
-            log.warn("No telemetry or attribute values is the SNMP response for device {}", sessionContext.getDeviceId());
-            return;
+            SnmpMapping mapping = mappings.get(oid);
+            processValue(mapping.getKey(), mapping.getDataType(), variableBinding.toValueString(), responses.get(spec));
         }
 
-        if (!telemetry.entrySet().isEmpty()) {
-            TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(telemetry);
-            transportService.process(sessionContext.getSessionInfo(), postTelemetryMsg, TransportServiceCallback.EMPTY);
-            log.debug("Posted telemetry for device {}: {}", sessionContext.getDeviceId(), telemetry);
+        if (responses.values().stream().allMatch(response -> response.entrySet().isEmpty())) {
+            log.debug("No values is the SNMP response for device {}. Request id: {}", sessionContext.getDeviceId(), responsePdu.getRequestID());
+            return;
         }
 
-        if (!attributes.entrySet().isEmpty()) {
-            TransportProtos.PostAttributeMsg postAttributesMsg = JsonConverter.convertToAttributesProto(attributes);
-            transportService.process(sessionContext.getSessionInfo(), postAttributesMsg, TransportServiceCallback.EMPTY);
-            log.debug("Posted attributes for device {}: {}", sessionContext.getDeviceId(), attributes);
-        }
+        responses.forEach((spec, response) -> {
+            Optional.ofNullable(responseProcessors.get(spec))
+                    .ifPresent(responseProcessor -> {
+                        responseProcessor.accept(response, sessionContext);
+                    });
+        });
 
         reportActivity(sessionContext.getSessionInfo());
     }
 
+    private void configureResponseProcessors() {
+        Stream.of(SnmpCommunicationSpec.TELEMETRY_QUERYING, SnmpCommunicationSpec.TELEMETRY_TRAPS_RECEIVING)
+                .forEach(telemetrySpec -> {
+                    responseProcessors.put(telemetrySpec, (response, sessionContext) -> {
+                        TransportProtos.PostTelemetryMsg postTelemetryMsg = JsonConverter.convertToTelemetryProto(response);
+                        transportService.process(sessionContext.getSessionInfo(), postTelemetryMsg, TransportServiceCallback.EMPTY);
+                        log.debug("Posted telemetry for device {}: {}", sessionContext.getDeviceId(), response);
+                    });
+                });
+
+        Stream.of(SnmpCommunicationSpec.CLIENT_ATTRIBUTES_QUERYING, SnmpCommunicationSpec.CLIENT_ATTRIBUTES_TRAPS_RECEIVING)
+                .forEach(clientAttributesSpec -> {
+                    responseProcessors.put(clientAttributesSpec, (response, sessionContext) -> {
+                        TransportProtos.PostAttributeMsg postAttributesMsg = JsonConverter.convertToAttributesProto(response);
+                        transportService.process(sessionContext.getSessionInfo(), postAttributesMsg, TransportServiceCallback.EMPTY);
+                        log.debug("Posted attributes for device {}: {}", sessionContext.getDeviceId(), response);
+                    });
+                });
+    }
+
     private void reportActivity(TransportProtos.SessionInfoProto sessionInfo) {
         transportService.process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder()
                 .setAttributeSubscription(false)
@@ -184,6 +336,8 @@ public class SnmpTransportService implements TbTransportService {
     }
 
     private void processValue(String key, DataType dataType, String value, JsonObject result) {
+        if (StringUtils.isEmpty(value)) return;
+
         switch (dataType) {
             case LONG:
                 result.addProperty(key, Long.parseLong(value));
@@ -199,6 +353,11 @@ public class SnmpTransportService implements TbTransportService {
         }
     }
 
+    @Override
+    public void processPdu(CommandResponderEvent event) {
+        processTrap(event);
+    }
+
     @Override
     public String getName() {
         return "SNMP";
@@ -207,11 +366,11 @@ public class SnmpTransportService implements TbTransportService {
     @PreDestroy
     public void shutdown() {
         log.info("Stopping SNMP transport!");
-        if (pollingExecutor != null) {
-            pollingExecutor.shutdownNow();
+        if (queryingExecutor != null) {
+            queryingExecutor.shutdownNow();
         }
-        if (snmpResponseProcessingExecutor != null) {
-            snmpResponseProcessingExecutor.shutdownNow();
+        if (responseProcessingExecutor != null) {
+            responseProcessingExecutor.shutdownNow();
         }
         if (snmp != null) {
             try {

common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java

@@ -18,18 +18,31 @@ package org.thingsboard.server.transport.snmp.session;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
+import org.snmp4j.AbstractTarget;
 import org.snmp4j.CommunityTarget;
 import org.snmp4j.Target;
+import org.snmp4j.UserTarget;
 import org.snmp4j.event.ResponseEvent;
 import org.snmp4j.event.ResponseListener;
+import org.snmp4j.security.AuthSHA;
+import org.snmp4j.security.PrivDES;
+import org.snmp4j.security.SecurityLevel;
+import org.snmp4j.security.SecurityModel;
+import org.snmp4j.security.SecurityModels;
+import org.snmp4j.security.USM;
+import org.snmp4j.security.UsmUser;
 import org.snmp4j.smi.GenericAddress;
+import org.snmp4j.smi.OID;
 import org.snmp4j.smi.OctetString;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.device.data.SnmpDeviceTransportConfiguration;
 import org.thingsboard.server.common.data.device.profile.SnmpDeviceProfileTransportConfiguration;
 import org.thingsboard.server.common.data.id.DeviceId;
+import org.thingsboard.server.common.data.transport.snmp.SnmpCommunicationSpec;
+import org.thingsboard.server.common.data.transport.snmp.SnmpProtocolVersion;
 import org.thingsboard.server.common.transport.SessionMsgListener;
+import org.thingsboard.server.common.transport.adaptor.JsonConverter;
 import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg;
@@ -40,8 +53,13 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseM
 import org.thingsboard.server.transport.snmp.SnmpTransportContext;
 import org.thingsboard.server.transport.snmp.service.SnmpTransportService;
 
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 @Slf4j
 public class DeviceSessionContext extends DeviceAwareSessionContext implements SessionMsgListener, ResponseListener {
@@ -64,11 +82,19 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
     @Setter
     private long previousRequestExecutedAt = 0;
     private final AtomicInteger msgIdSeq = new AtomicInteger(0);
+    @Getter
     private boolean isActive = true;
+    private final String snmpUnderlyingProtocol;
+
+    @Getter
+    @Setter
+    private List<ScheduledFuture<?>> queryingTasks = new LinkedList<>();
 
-    public DeviceSessionContext(Device device, DeviceProfile deviceProfile,
-                                String token, SnmpDeviceTransportConfiguration deviceTransportConfiguration,
-                                SnmpTransportContext snmpTransportContext, SnmpTransportService snmpTransportService) {
+    public DeviceSessionContext(Device device, DeviceProfile deviceProfile, String token,
+                                SnmpDeviceProfileTransportConfiguration profileTransportConfiguration,
+                                SnmpDeviceTransportConfiguration deviceTransportConfiguration,
+                                SnmpTransportContext snmpTransportContext, SnmpTransportService snmpTransportService,
+                                String snmpUnderlyingProtocol) {
         super(UUID.randomUUID());
         super.setDeviceId(device.getId());
         super.setDeviceProfile(deviceProfile);
@@ -78,10 +104,11 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
         this.snmpTransportContext = snmpTransportContext;
         this.snmpTransportService = snmpTransportService;
 
-        this.profileTransportConfiguration = (SnmpDeviceProfileTransportConfiguration) deviceProfile.getProfileData().getTransportConfiguration();
+        this.profileTransportConfiguration = profileTransportConfiguration;
         this.deviceTransportConfiguration = deviceTransportConfiguration;
 
-        initTarget(this.profileTransportConfiguration, this.deviceTransportConfiguration);
+        this.snmpUnderlyingProtocol = snmpUnderlyingProtocol;
+        initializeTarget(profileTransportConfiguration, deviceTransportConfiguration);
     }
 
     @Override
@@ -100,20 +127,62 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
     @Override
     public void onResponse(ResponseEvent event) {
         if (isActive) {
-            snmpTransportService.onNewDeviceResponse(event, this);
+            snmpTransportService.processResponseEvent(this, event);
         }
     }
 
-    public void initTarget(SnmpDeviceProfileTransportConfiguration profileTransportConfig, SnmpDeviceTransportConfiguration deviceTransportConfig) {
+    public void initializeTarget(SnmpDeviceProfileTransportConfiguration profileTransportConfig, SnmpDeviceTransportConfiguration deviceTransportConfig) {
         log.trace("Initializing target for SNMP session of device {}", device);
-        CommunityTarget communityTarget = new CommunityTarget();
-        communityTarget.setAddress(GenericAddress.parse(GenericAddress.TYPE_UDP + ":" + deviceTransportConfig.getAddress() + "/" + deviceTransportConfig.getPort()));
-        communityTarget.setVersion(deviceTransportConfig.getProtocolVersion().getCode());
-        communityTarget.setCommunity(new OctetString(deviceTransportConfig.getCommunity()));
-        communityTarget.setTimeout(profileTransportConfig.getTimeoutMs());
-        communityTarget.setRetries(profileTransportConfig.getRetries());
-        this.target = communityTarget;
-        log.info("SNMP target initialized: {}", this.target);
+
+        AbstractTarget target;
+
+        SnmpProtocolVersion protocolVersion = deviceTransportConfig.getProtocolVersion();
+        switch (protocolVersion) {
+            case V1:
+                CommunityTarget communityTargetV1 = new CommunityTarget();
+                communityTargetV1.setSecurityModel(SecurityModel.SECURITY_MODEL_SNMPv1);
+                communityTargetV1.setSecurityLevel(SecurityLevel.NOAUTH_NOPRIV);
+                communityTargetV1.setCommunity(new OctetString(deviceTransportConfig.getSecurityName()));
+                target = communityTargetV1;
+                break;
+            case V2C:
+                CommunityTarget communityTargetV2 = new CommunityTarget();
+                communityTargetV2.setSecurityModel(SecurityModel.SECURITY_MODEL_SNMPv2c);
+                communityTargetV2.setSecurityLevel(SecurityLevel.NOAUTH_NOPRIV);
+                communityTargetV2.setCommunity(new OctetString(deviceTransportConfig.getSecurityName()));
+                target = communityTargetV2;
+                break;
+            case V3:
+                USM usm = new USM();
+                SecurityModels.getInstance().addSecurityModel(usm);
+
+                OctetString securityName = new OctetString(deviceTransportConfig.getSecurityName());
+                OctetString authenticationPassphrase = new OctetString(deviceTransportConfig.getAuthenticationPassphrase());
+                OctetString privacyPassphrase = new OctetString(deviceTransportConfig.getPrivacyPassphrase());
+
+                OID authenticationProtocol = AuthSHA.ID;
+                OID privacyProtocol = PrivDES.ID; // FIXME: to config
+
+                UsmUser user = new UsmUser(securityName, authenticationProtocol, authenticationPassphrase, privacyProtocol, privacyPassphrase);
+                snmpTransportService.getSnmp().getUSM().addUser(user);
+
+                UserTarget userTarget = new UserTarget();
+                userTarget.setSecurityName(securityName);
+                userTarget.setSecurityLevel(SecurityLevel.AUTH_PRIV);
+
+                target = userTarget;
+                break;
+            default:
+                throw new UnsupportedOperationException("SNMP protocol version " + protocolVersion + " is not supported");
+        }
+
+        target.setAddress(GenericAddress.parse(snmpUnderlyingProtocol + ":" + deviceTransportConfig.getAddress() + "/" + deviceTransportConfig.getPort()));
+        target.setTimeout(profileTransportConfig.getTimeoutMs());
+        target.setRetries(profileTransportConfig.getRetries());
+        target.setVersion(protocolVersion.getCode());
+
+        this.target = target;
+        log.info("SNMP target initialized: {}", target);
     }
 
     public void close() {
@@ -135,6 +204,21 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
 
     @Override
     public void onAttributeUpdate(AttributeUpdateNotificationMsg attributeUpdateNotification) {
+        profileTransportConfiguration.getCommunicationConfigs().stream()
+                .filter(config -> config.getSpec() == SnmpCommunicationSpec.SHARED_ATTRIBUTES_SETTING)
+                .findFirst()
+                .ifPresent(communicationConfig -> {
+                    Map<String, String> sharedAttributes = JsonConverter.toJson(attributeUpdateNotification).entrySet().stream()
+                            .collect(Collectors.toMap(
+                                    Map.Entry::getKey,
+                                    entry -> entry.getValue().isJsonPrimitive() ? entry.getValue().getAsString() : entry.getValue().toString()
+                            ));
+                    try {
+                        snmpTransportService.sendRequest(this, communicationConfig, sharedAttributes);
+                    } catch (Exception e) {
+                        log.error("Failed to send request with shared attributes to SNMP device {}: {}", getDeviceId(), e.getMessage());
+                    }
+                });
     }
 
     @Override

common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV2.java

@@ -0,0 +1,222 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.snmp;
+
+import org.snmp4j.CommandResponderEvent;
+import org.snmp4j.CommunityTarget;
+import org.snmp4j.PDU;
+import org.snmp4j.Snmp;
+import org.snmp4j.Target;
+import org.snmp4j.TransportMapping;
+import org.snmp4j.agent.BaseAgent;
+import org.snmp4j.agent.CommandProcessor;
+import org.snmp4j.agent.DuplicateRegistrationException;
+import org.snmp4j.agent.MOGroup;
+import org.snmp4j.agent.ManagedObject;
+import org.snmp4j.agent.mo.MOAccessImpl;
+import org.snmp4j.agent.mo.MOScalar;
+import org.snmp4j.agent.mo.snmp.RowStatus;
+import org.snmp4j.agent.mo.snmp.SnmpCommunityMIB;
+import org.snmp4j.agent.mo.snmp.SnmpNotificationMIB;
+import org.snmp4j.agent.mo.snmp.SnmpTargetMIB;
+import org.snmp4j.agent.mo.snmp.StorageType;
+import org.snmp4j.agent.mo.snmp.VacmMIB;
+import org.snmp4j.agent.security.MutableVACM;
+import org.snmp4j.mp.MPv3;
+import org.snmp4j.mp.SnmpConstants;
+import org.snmp4j.security.SecurityLevel;
+import org.snmp4j.security.SecurityModel;
+import org.snmp4j.security.USM;
+import org.snmp4j.smi.Address;
+import org.snmp4j.smi.GenericAddress;
+import org.snmp4j.smi.Integer32;
+import org.snmp4j.smi.OID;
+import org.snmp4j.smi.OctetString;
+import org.snmp4j.smi.UdpAddress;
+import org.snmp4j.smi.Variable;
+import org.snmp4j.smi.VariableBinding;
+import org.snmp4j.transport.TransportMappings;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Scanner;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+public class SnmpDeviceSimulatorV2 extends BaseAgent {
+
+    public static class RequestProcessor extends CommandProcessor {
+        private final Consumer<CommandResponderEvent> processor;
+
+        public RequestProcessor(Consumer<CommandResponderEvent> processor) {
+            super(new OctetString(MPv3.createLocalEngineID()));
+            this.processor = processor;
+        }
+
+        @Override
+        public void processPdu(CommandResponderEvent event) {
+            processor.accept(event);
+        }
+    }
+
+    public static void main(String[] args) throws IOException {
+        SnmpDeviceSimulatorV2 device = new SnmpDeviceSimulatorV2(1610, "public");
+
+        device.start();
+        device.setUpMappings(Map.of(
+                ".1.3.6.1.2.1.1.1.50", "12",
+                ".1.3.6.1.2.1.2.1.52", "56",
+                ".1.3.6.1.2.1.3.1.54", "yes",
+                ".1.3.6.1.2.1.7.1.58", ""
+        ));
+
+
+//        while (true) {
+//            new Scanner(System.in).nextLine();
+//            device.sendTrap("127.0.0.1", 1062, Map.of(".1.3.6.1.2.87.1.56", "12"));
+//            System.out.println("sent");
+//        }
+
+//        Snmp snmp = new Snmp(device.transportMappings[0]);
+//        device.snmp.addCommandResponder(event -> {
+//            System.out.println(event);
+//        });
+
+        new Scanner(System.in).nextLine();
+    }
+
+
+    private final Target target;
+    private final Address address;
+    private Snmp snmp;
+
+    private final String password;
+
+    public SnmpDeviceSimulatorV2(int port, String password) throws IOException {
+        super(new File("conf.agent"), new File("bootCounter.agent"), new RequestProcessor(event -> {
+            System.out.println("aboba");
+            ((Snmp) event.getSource()).cancel(event.getPDU(), event1 -> System.out.println("canceled"));
+        }));
+        CommunityTarget target = new CommunityTarget();
+        target.setCommunity(new OctetString(password));
+        this.address = GenericAddress.parse("udp:0.0.0.0/" + port);
+        target.setAddress(address);
+        target.setRetries(2);
+        target.setTimeout(1500);
+        target.setVersion(SnmpConstants.version2c);
+        this.target = target;
+        this.password = password;
+    }
+
+    public void start() throws IOException {
+        init();
+        addShutdownHook();
+        getServer().addContext(new OctetString("public"));
+        finishInit();
+        run();
+        sendColdStartNotification();
+        snmp = new Snmp(transportMappings[0]);
+    }
+
+    public void setUpMappings(Map<String, String> oidToResponseMappings) {
+        unregisterManagedObject(getSnmpv2MIB());
+        oidToResponseMappings.forEach((oid, response) -> {
+            registerManagedObject(new MOScalar<>(new OID(oid), MOAccessImpl.ACCESS_READ_WRITE, new OctetString(response)));
+        });
+    }
+
+    public void sendTrap(String host, int port, Map<String, String> values) throws IOException {
+        PDU pdu = new PDU();
+        pdu.addAll(values.entrySet().stream()
+                .map(entry -> new VariableBinding(new OID(entry.getKey()), new OctetString(entry.getValue())))
+                .collect(Collectors.toList()));
+        pdu.setType(PDU.TRAP);
+
+        CommunityTarget remoteTarget = (CommunityTarget) getTarget().clone();
+        remoteTarget.setAddress(new UdpAddress(host + "/" + port));
+
+        snmp.send(pdu, remoteTarget);
+    }
+
+    @Override
+    protected void registerManagedObjects() {
+    }
+
+    protected void registerManagedObject(ManagedObject mo) {
+        try {
+            server.register(mo, null);
+        } catch (DuplicateRegistrationException ex) {
+            throw new RuntimeException(ex);
+        }
+    }
+
+    protected void unregisterManagedObject(MOGroup moGroup) {
+        moGroup.unregisterMOs(server, getContext(moGroup));
+    }
+
+    @Override
+    protected void addNotificationTargets(SnmpTargetMIB targetMIB,
+                                          SnmpNotificationMIB notificationMIB) {
+    }
+
+    @Override
+    protected void addViews(VacmMIB vacm) {
+        vacm.addGroup(SecurityModel.SECURITY_MODEL_SNMPv2c, new OctetString(
+                        "cpublic"), new OctetString("v1v2group"),
+                StorageType.nonVolatile);
+
+        vacm.addAccess(new OctetString("v1v2group"), new OctetString("public"),
+                SecurityModel.SECURITY_MODEL_ANY, SecurityLevel.NOAUTH_NOPRIV,
+                MutableVACM.VACM_MATCH_EXACT, new OctetString("fullReadView"),
+                new OctetString("fullWriteView"), new OctetString(
+                        "fullNotifyView"), StorageType.nonVolatile);
+
+        vacm.addViewTreeFamily(new OctetString("fullReadView"), new OID("1.3"),
+                new OctetString(), VacmMIB.vacmViewIncluded,
+                StorageType.nonVolatile);
+    }
+
+    protected void addUsmUser(USM usm) {
+    }
+
+    protected void initTransportMappings() {
+        transportMappings = new TransportMapping[]{TransportMappings.getInstance().createTransportMapping(address)};
+    }
+
+    protected void unregisterManagedObjects() {
+    }
+
+    protected void addCommunities(SnmpCommunityMIB communityMIB) {
+        Variable[] com2sec = new Variable[]{
+                new OctetString("public"),
+                new OctetString("cpublic"),
+                getAgent().getContextEngineID(),
+                new OctetString("public"),
+                new OctetString(),
+                new Integer32(StorageType.nonVolatile),
+                new Integer32(RowStatus.active)
+        };
+        SnmpCommunityMIB.SnmpCommunityEntryRow row = communityMIB.getSnmpCommunityEntry().createRow(
+                new OctetString("public2public").toSubIndex(true), com2sec);
+        communityMIB.getSnmpCommunityEntry().addRow(row);
+    }
+
+    public Target getTarget() {
+        return target;
+    }
+
+}

common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpDeviceSimulatorV3.java

@@ -0,0 +1,52 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.snmp;
+
+import org.snmp4j.UserTarget;
+import org.snmp4j.security.AuthSHA;
+import org.snmp4j.security.PrivDES;
+import org.snmp4j.security.SecurityLevel;
+import org.snmp4j.security.SecurityModels;
+import org.snmp4j.security.USM;
+import org.snmp4j.security.UsmUser;
+import org.snmp4j.smi.OID;
+import org.snmp4j.smi.OctetString;
+
+import java.io.IOException;
+
+public class SnmpDeviceSimulatorV3 extends SnmpDeviceSimulatorV2 {
+    public SnmpDeviceSimulatorV3(int port, String securityName, String authenticationPassphrase, String privacyPassphrase) throws IOException {
+        super(12, null);
+//        super(new File("conf.agent"), new File("bootCounter.agent"));
+
+        USM usm = new USM();
+        SecurityModels.getInstance().addSecurityModel(usm);
+
+        OID authenticationProtocol = AuthSHA.ID;
+        OID privacyProtocol = PrivDES.ID;
+
+        UsmUser user = new UsmUser(new OctetString(securityName), authenticationProtocol, new OctetString(authenticationPassphrase), privacyProtocol, new OctetString(privacyPassphrase));
+    }
+
+    public void initV3(UsmUser user, String securityName) {
+//        snmp.getUSM().addUser(user);
+
+        UserTarget userTarget = new UserTarget();
+        userTarget.setSecurityName(new OctetString(securityName));
+        userTarget.setSecurityLevel(SecurityLevel.AUTH_PRIV);
+    }
+
+}

common/transport/snmp/src/test/java/org/thingsboard/server/transport/snmp/SnmpTest.java

@@ -0,0 +1,49 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.transport.snmp;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Scanner;
+
+public class SnmpTest {
+    public static void main(String[] args) throws IOException {
+        SnmpDeviceSimulatorV2 device = new SnmpDeviceSimulatorV2(1610, "public");
+
+        device.start();
+        device.setUpMappings(Map.of(
+                ".1.3.6.1.2.1.1.1.50", "12",
+                ".1.3.6.1.2.1.2.1.52", "56",
+                ".1.3.6.1.2.1.3.1.54", "yes",
+                ".1.3.6.1.2.1.7.1.58", ""
+        ));
+
+
+//        while (true) {
+//            new Scanner(System.in).nextLine();
+//            device.sendTrap("127.0.0.1", 1062, Map.of(".1.3.6.1.2.87.1.56", "12"));
+//            System.out.println("sent");
+//        }
+
+//        Snmp snmp = new Snmp(device.transportMappings[0]);
+//        device.snmp.addCommandResponder(event -> {
+//            System.out.println(event);
+//        });
+
+        new Scanner(System.in).nextLine();
+    }
+
+}

common/transport/snmp/src/test/resources/snmp-device-profile-transport-config.json

@@ -0,0 +1,53 @@
+{
+  "timeoutMs": 500,
+  "retries": 0,
+  "communicationConfigs": [
+    {
+      "spec": "TELEMETRY_QUERYING",
+      "queryingFrequencyMs": 3000,
+      "mappings": [
+        {
+          "oid": ".1.3.6.1.2.1.1.1.50",
+          "key": "temperature",
+          "dataType": "LONG"
+        },
+        {
+          "oid": ".1.3.6.1.2.1.2.1.52",
+          "key": "humidity",
+          "dataType": "DOUBLE"
+        }
+      ]
+    },
+    {
+      "spec": "CLIENT_ATTRIBUTES_QUERYING",
+      "queryingFrequencyMs": 5000,
+      "mappings": [
+        {
+          "oid": ".1.3.6.1.2.1.3.1.54",
+          "key": "isCool",
+          "dataType": "STRING"
+        }
+      ]
+    },
+    {
+      "spec": "SHARED_ATTRIBUTES_SETTING",
+      "mappings": [
+        {
+          "oid": ".1.3.6.1.2.1.7.1.58",
+          "key": "shared",
+          "dataType": "STRING"
+        }
+      ]
+    },
+    {
+      "spec": "TELEMETRY_TRAPS_RECEIVING",
+      "mappings": [
+        {
+          "oid": ".1.3.6.1.2.8.7.1.56",
+          "key": "temperature_trap",
+          "dataType": "LONG"
+        }
+      ]
+    }
+  ]
+}

common/transport/snmp/src/test/resources/snmp-device-transport-config.json

@@ -0,0 +1,6 @@
+{
+  "address": "127.0.0.1",
+  "port": 1610,
+  "password": "public",
+  "protocolVersion": "V2C"
+}

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportResourceCache.java

@@ -15,13 +15,15 @@
  */
 package org.thingsboard.server.common.transport;
 
-import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.ResourceType;
+import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.id.TenantId;
 
+import java.util.Optional;
+
 public interface TransportResourceCache {
 
-    TbResource get(TenantId tenantId, ResourceType resourceType, String resourceId);
+    Optional<TbResource> get(TenantId tenantId, ResourceType resourceType, String resourceId);
 
     void update(TenantId tenantId, ResourceType resourceType, String resourceI);
 

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java

@@ -285,7 +285,7 @@ public class JsonConverter {
         return result;
     }
 
-    public static JsonElement toJson(AttributeUpdateNotificationMsg payload) {
+    public static JsonObject toJson(AttributeUpdateNotificationMsg payload) {
         JsonObject result = new JsonObject();
         if (payload.getSharedUpdatedCount() > 0) {
             payload.getSharedUpdatedList().forEach(addToObjectFromProto(result));

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportResourceCache.java

@@ -19,8 +19,8 @@ import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
-import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.ResourceType;
+import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.transport.TransportResourceCache;
 import org.thingsboard.server.common.transport.TransportService;
@@ -53,7 +53,7 @@ public class DefaultTransportResourceCache implements TransportResourceCache {
     }
 
     @Override
-    public TbResource get(TenantId tenantId, ResourceType resourceType, String resourceKey) {
+    public Optional<TbResource> get(TenantId tenantId, ResourceType resourceType, String resourceKey) {
         ResourceCompositeKey compositeKey = new ResourceCompositeKey(tenantId, resourceType, resourceKey);
         TbResource resource;
 
@@ -79,7 +79,7 @@ public class DefaultTransportResourceCache implements TransportResourceCache {
             }
         }
 
-        return resource;
+        return Optional.ofNullable(resource);
     }
 
     private TbResource fetchResource(ResourceCompositeKey compositeKey) {

dao/src/main/java/org/thingsboard/server/dao/resource/BaseTbResourceService.java

@@ -87,7 +87,7 @@ public class BaseTbResourceService implements TbResourceService {
                 String resourceKey = objectModel.id + LWM2M_SEPARATOR_KEY + objectModel.getVersion();
                 String name = objectModel.name;
                 resource.setResourceKey(resourceKey);
-                resource.setTitle(name);
+                resource.setTitle(name + " id=" +objectModel.id + " v" + objectModel.getVersion());
                 resource.setSearchText(resourceKey + LWM2M_SEPARATOR_SEARCH_TEXT + name);
             } else {
                 throw new DataValidationException(String.format("Could not parse the XML of objectModel with name %s", resource.getSearchText()));

dao/src/test/java/org/thingsboard/server/dao/service/BaseTbResourceServiceTest.java

@@ -123,7 +123,7 @@ public abstract class BaseTbResourceServiceTest extends AbstractServiceTest {
         Assert.assertNotNull(savedResource.getId());
         Assert.assertTrue(savedResource.getCreatedTime() > 0);
         Assert.assertEquals(resource.getTenantId(), savedResource.getTenantId());
-        Assert.assertEquals("My first resource", savedResource.getTitle());
+        Assert.assertEquals("My first resource id=0 v1.0", savedResource.getTitle());
         Assert.assertEquals("0_1.0", savedResource.getResourceKey());
         Assert.assertEquals(resource.getData(), savedResource.getData());
 

transport/snmp/src/main/resources/device-data-config.json

@@ -1,9 +0,0 @@
-{
-  "deviceName": "Thermostat T1",
-  "snmpConfig": {
-    "address": "192.168.1.2",
-    "port": 161,
-    "community": "U5J=$HWj6f@7",
-    "protocolVersion": "v2c"
-  }
-}

transport/snmp/src/main/resources/device-profile-config.json

@@ -1,21 +0,0 @@
-{
-  "poolPeriodMs": 10000,
-  "timeoutMs": 5000,
-  "retries": 5,
-  "attributes": [
-    {
-      "key": "snmpNodeManagerEmail",
-      "type": "STRING",
-      "method": "get",
-      "oid": ".1.3.6.1.2.1.1.4.0"
-    }
-  ],
-  "telemetry": [
-    {
-      "key": "snmpNodeSysUpTime",
-      "type": "LONG",
-      "method": "get",
-      "oid": ".1.3.6.1.2.1.1.3.0"
-    }
-  ]
-}

transport/snmp/src/main/resources/tb-snmp-transport.yml

@@ -189,6 +189,11 @@ transport:
   # Local LwM2M transport parameters
   snmp:
     enabled: "${SNMP_ENABLED:true}"
+    response_processing:
+      # parallelism level for executor (workStealingPool) that is responsible for handling responses from SNMP devices
+      parallelism_level: "${SNMP_RESPONSE_PROCESSING_PARALLELISM_LEVEL:20}"
+    # to configure SNMP to work over UDP or TCP
+    underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}"
 
 queue:
   type: "${TB_QUEUE_TYPE:in-memory}" # in-memory or kafka (Apache Kafka) or aws-sqs (AWS SQS) or pubsub (PubSub) or service-bus (Azure Service Bus) or rabbitmq (RabbitMQ)